Money laundering poses a significant threat to global financial systems by disguising illicit funds as legitimate transactions. This project develops a scalable anti-money laundering (AML) detection system using Apache Spark and Hadoop Distributed File System (HDFS) to analyze a large financial transaction dataset. Stored on HDFS, the dataset encompasses diverse payment types, currencies, and geographic regions, with challenges such as severe class imbalance due to the rarity of fraudulent transactions. Utilizing PySpark, the project implements a machine learning pipeline that addresses these issues through feature engineering, generating temporal, behavioral, and geographic risk indicators like cross-border flags and sender activity patterns. A custom Synthetic Minority Oversampling Technique (SMOTE) is applied to balance the dataset, improving model sensitivity to fraud. Multiple machine learning algorithms, including Logistic Regression, Decision Tree, Random Forest, Gradient-Boosted Trees, Naive Bayes, Linear SVM, and Multilayer Perceptron, are trained and evaluated, with a retrained Decision Tree yielding superior performance. The pipeline leverages Spark’s distributed processing for scalability, ensuring adaptability to larger datasets. Visualizations, such as confusion matrices, underscore model effectiveness. This work delivers a robust, scalable AML solution that minimizes false negatives, supports regulatory compliance, and provides financial institutions with a blueprint for big data-driven fraud detection in the digital economy.
Note - The dataset is stored in my HDFS in the name SAML-D.csv and kaggle reference provided below in the reference list (Reference -1) and all 3 components are in the same notebook have also submitted a html along with notebook file of this same document for clarity.
Money laundering hides illegal funds within legitimate transactions, threatening global finance. This project builds a scalable anti-money laundering (AML) detection system using Apache Spark and Hadoop Distributed File System (HDFS). The large financial transaction dataset, stored across HDFS nodes, is split and replicated for fault tolerance and fast access. Each node handles data chunks, enabling parallel processing. Using PySpark, we create a machine learning pipeline with feature engineering to extract risk indicators like cross-border flags. A custom Synthetic Minority Oversampling Technique (SMOTE) balances rare fraud cases. Multiple algorithms are tested, with Spark distributing computations across nodes for speed. This ensures scalability, reduces false negatives, and supports compliance for financial institutions.
The SAML-D.csv dataset, sourced from Kaggle (Reference-1), is a 996 MB financial transaction dataset containing 9,504,852 records, stored on Hadoop Distributed File System for distributed processing. It includes 12 columns: Time (transaction time), Date (timestamp), Sender_account and Receiver_account (account IDs), Amount (transaction value), Payment_currency and Received_currency (e.g., UK pounds, Dirham), Sender_bank_location and Receiver_bank_location (e.g., UK, UAE), Payment_type (e.g., Cash Deposit, Cross-border), Is_laundering (binary label, with 9,873 or 0.1% fraudulent transactions), and Laundering_type (transaction pattern). Derived features include Hour, Day_of_Week, Sender_Txn_Count, Is_Cross_Border, Log_Amount, Avg_Amount, and Txn_Frequency, enhancing fraud detection. The dataset’s severe class imbalance, with 9,494,979 non-fraudulent transactions, is mitigated, and Amount outliers are capped using the interquartile range method for robust analysis in scalable machine learning pipelines.
The primary aim of this project is to develop a scalable and efficient anti-money laundering (AML) detection system capable of processing large-scale financial transaction data to identify illicit activities with high accuracy. Leveraging Apache Spark and Hadoop Distributed File System (HDFS), the system processes the 996 MB SAML-D.csv dataset, containing 9.5 million records, stored across HDFS nodes for fault tolerance and parallel access. PySpark is used to build a machine learning pipeline that includes feature engineering, class imbalance mitigation via Synthetic Minority Oversampling Technique (SMOTE), and model training with algorithms like Decision Tree, optimized for distributed computing. The objective is to achieve high precision and recall, minimizing false negatives to support regulatory compliance, using Spark MLlib for scalable model development and Plotly for visualizing results like confusion matrices.
Specific objectives include implementing feature engineering to derive risk indicators (e.g., Sender_Txn_Count, Is_Cross_Border) using PySpark’s DataFrame API, ensuring scalability on HDFS-stored data. The project addresses class imbalance (0.1% fraud) by applying SMOTE in PySpark to balance the dataset, enhancing model sensitivity. Multiple models, including Logistic Regression and Random Forest, are trained and evaluated on a distributed Spark cluster, with the Decision Tree retrained (maxDepth=15) to optimize performance. HDFS ensures data reliability, while PySpark’s distributed processing handles large-scale computations. Pandas and Plotly are used for exploratory analysis and visualizations, such as bar charts, to compare model metrics. The system aims to provide a robust AML framework for financial institutions, scalable to larger datasets, supporting real-time fraud detection and FATF compliance.
This project uses powerful technologies to detect money laundering in a large financial dataset. The dataset, stored on the Hadoop Distributed File System (HDFS), is split across multiple computers for safe and fast access. Apache Spark, a tool for handling big data, processes the dataset quickly by dividing tasks across these computers. PySpark, the Python version of Spark, is used to write simple code for analyzing the data and building machine learning models. master("yarn"): This indicates that YARN (Yet Another Resource Negotiator) is being used as the resource manager, meaning Spark jobs are distributed and run across a cluster config("spark.hadoop.fs.defaultFS", "hdfs://dsm-master:9000") \: This configuration tells Spark to read and write data from the Hadoop file system (HDFS), with namenode:8020 acting as the central coordinator for file system metadata.
Spark MLlib, a library in Spark, provides tools to create and test models like Decision Trees and Logistic Regression. These technologies work together to handle millions of transactions efficiently. They ensure the system can grow to manage even larger datasets.For visualizations, Python libraries like Pandas and Plotly are used to create charts and graphs. Pandas helps organize small data chunks for plotting, while Plotly makes interactive visuals like bar charts and confusion matrices to show results clearly. HDFS and Spark make the project scalable, meaning it can handle more data as needed. MLlib’s models are designed to work on distributed systems, ensuring fast and accurate fraud detection. Python ties everything together, making coding and visualization straightforward. This setup allows financial institutions to detect money laundering effectively while meeting regulatory needs. It’s a strong, flexible system for big data analysis.
The project combats money laundering, a global threat hiding illicit funds in legitimate transactions, costing billions and fueling crimes like drug trafficking and terrorism. It develops a scalable anti-money laundering (AML) system vital for banks and regulators to meet Financial Action Task Force (FATF) compliance. Using Hadoop Distributed File System (HDFS), the 996 MB dataset with 9,504,852 transactions is stored across nodes, enabling parallel access and fault tolerance. Apache Spark processes this data via PySpark, using DataFrame APIs for real-time analysis of features like Amount and Payment_type. Spark MLlib builds models (e.g., Decision Tree, maxDepth=15) on 7.68 million records, with Synthetic Minority Oversampling Technique (SMOTE) addressing 0.1% fraud imbalance. Pandas and Plotly create visualizations like confusion matrices. This distributed pipeline ensures scalability, achieves 0.999 accuracy, minimizes false negatives, saves banks from fines, and disrupts crime. It advances big data research, offering a scalable AML detection framework.
In PySpark, this code first creates Spark DataFrames for project steps (nodes) and their connections (edges). It converts them to Pandas DataFrames so Plotly can use them to draw the flowchart. Each phase of the AML pipeline is shown as a labeled box, connected by arrows to show the order. After plotting, it stops the PySpark session to free resources.
nodes_data=[(0,"1. Data Acquisition & Storage","Load 996 MB SAML-D.csv (9.5M records) into Spark DataFrame from HDFS."),(1,"2. Exploratory Data Analysis","Compute stats (0.1% fraud). Analyze payment types, sender amounts."),(2,"3. Preprocessing & Feature Engineering","Cap Amount outliers. Derive Hour, Sender_Txn_Count, Is_Cross_Border."),(3,"4. Addressing Class Imbalance","Apply SMOTE to balance 9,873 fraud vs. 9.5M non-fraud records."),(4,"5. Model Development & Training","Split data (80/20). Train Decision Tree with Spark MLlib pipeline."),(5,"6. Model Evaluation & Optimization","Evaluate (accuracy: 0.999, recall: 0.905). Retrain Decision Tree.")]
edges_data=[(0,1),(1,2),(2,3),(3,4),(4,5)]
nodes_df=spark.createDataFrame(nodes_data,["id","label","description"])
edges_df=spark.createDataFrame(edges_data,["source","target"])
nodes_pd=nodes_df.toPandas()
edges_pd=edges_df.toPandas()
fig=go.Figure()
node_x=[0.5]*len(nodes_pd)
node_y=[1-i*0.15 for i in range(len(nodes_pd))]
edge_x=[];edge_y=[]
for _,e in edges_pd.iterrows():edge_x+=[node_x[e['source']],node_x[e['target']],None];edge_y+=[node_y[e['source']]-0.02,node_y[e['target']]+0.02,None]
fig.add_trace(go.Scatter(x=edge_x,y=edge_y,line=dict(width=2,color='black'),hoverinfo='none',mode='lines'))
annotations=[dict(x=node_x[i],y=node_y[i],xref="paper",yref="paper",text=f"<b>{r['label']}</b><br>{r['description']}",showarrow=False,align="center",bgcolor="lightblue",bordercolor="black",borderwidth=1,width=250,font=dict(size=10),xanchor="center",yanchor="middle")for i,r in nodes_pd.iterrows()]
fig.add_trace(go.Scatter(x=node_x,y=node_y,mode='markers',marker=dict(size=20,color='lightblue',line=dict(width=2,color='black')),hoverinfo='none'))
fig.update_layout(title="AML Detection Project Methodology Flowchart",showlegend=False,xaxis=dict(showgrid=False,zeroline=False,showticklabels=False),yaxis=dict(showgrid=False,zeroline=False,showticklabels=False),plot_bgcolor='white',height=800,width=800,annotations=annotations)
fig.show()
spark.stop()
The project starts by getting the SAML-D.csv dataset, which has 9.5 million financial records and is 996 MB in size. This dataset is stored on the Hadoop Distributed File System (HDFS), a system that spreads data across many computers for fast and safe access. HDFS splits the dataset into smaller pieces, copying them to different nodes to prevent data loss. This setup lets Apache Spark read the data quickly using PySpark code. The dataset has 12 columns, like Amount, Payment_type, and Is_laundering, which show transaction details. Storing it on HDFS ensures the project can handle large datasets, even bigger than this one. The data is loaded into a Spark DataFrame for further work. This step builds a strong base for the project’s analysis.
Next, the project explores the dataset to understand its patterns. Using PySpark, we calculate numbers like the average Amount (7,343.75) and find no missing data. Only 9,873 records (0.1%) are fraud, showing a big class imbalance. Payment types include Credit Card (~21%), Cross-border (9.83%), and Cash Deposit (2.37%). We find top senders, like account 1534642148, with 15 million in transactions. PySpark groups data, and Pandas with Plotly creates charts, like bar graphs for payment types and histograms for fraud. These visuals show what’s common and what’s risky. This step helps decide which features to use for catching fraud.
To prepare the data, we clean and add new features using PySpark. We cap extreme Amount values (3.73 to 22,783.95) to avoid errors. New features include Hour (from Time), Day_of_Week (from Date), and Sender_Txn_Count (transactions per sender). Is_Cross_Border flags if sender and receiver locations differ (9.83% of cases). Log_Amount normalizes Amount, and Avg_Amount tracks sender averages. A custom Synthetic Minority Oversampling Technique (SMOTE) balances fraud (9,873) and non-fraud (9.5 million) records by creating synthetic fraud data. Categorical columns like Payment_type are numbered for models. These steps make the data ready for accurate fraud detection.
Finally, we build and test machine learning models with Spark MLlib. The dataset splits into training (80%, 7.7 million records) and test (20%, 1.9 million) sets. Models like Decision Tree, Logistic Regression, and Random Forest are trained on features like Sender_Txn_Count. The Decision Tree (maxDepth=15) performs best, with 0.999 accuracy. Metrics like precision, recall, and ROC AUC are calculated using Spark evaluators. Plotly creates confusion matrices and bar charts to compare models. The system scales with Spark’s distributed computing, handling large data easily. This step finds the best model to catch fraud and save banks from fines.
RDDs (Resilient Distributed Datasets) and DataFrames are core PySpark abstractions for processing the SAML-D.csv dataset. RDDs are low-level, distributed collections of objects, offering fine-grained control over data transformations (e.g., map, reduceByKey). The project uses RDDs to compute total transaction amounts per sender (sender_amount_rdd), leveraging their flexibility for custom operations. However, RDDs lack built-in optimization and schema, requiring manual coding. DataFrames, higher-level abstractions, organize data into named columns with a schema, resembling SQL tables. The project extensively uses DataFrames for loading data (spark.read.csv), feature engineering (df.withColumn), and ML pipelines, benefiting from Catalyst optimizer for efficient query execution. DataFrames simplify operations and integrate with Spark MLlib, unlike RDDs, which are less suited for structured data. For scalability, DataFrames are preferred for most tasks, while RDDs handle specific low-level computations.
The first idea is that transactions with many transfers from one account (Sender_Txn_Count) or those sent across countries (Is_Cross_Border=1) are more likely to be fraud. These could show criminals hiding money through complex patterns. The second idea is that fraud happens more at odd times, like late at night (Hour) or on certain days (Day_of_Week). To test these, we use PySpark to process the large dataset (SAML-D.csv) across nodes, ensuring fast analysis. HDFS stores the data, splitting it across nodes for quick access. This helps banks find fraud quickly. PySpark analyzes features like Sender_Txn_Count or Log_Amount to find key fraud patterns. Nodes in the cluster run tasks in parallel, speeding up processing. HDFS ensures data is safe and scalable for bigger datasets. The project’s goal is to help banks catch money laundering in real-time, saving them from big fines and stopping criminals. These ideas follow rules from the Financial Action Task Force (FATF) for fighting money laundering. Testing these patterns ensures banks stay safe and legal.
This code is used to prepare for data analysis and machine learning using PySpark. It starts by importing SparkSession, which is needed to create and run a Spark application. The Window and Row modules help with advanced data operations, such as ranking or working with individual rows. A group of useful functions like col, when, count, mean, and stddev are imported to manipulate and explore data easily. These functions are also imported using a short name F, so you can write F.mean() instead of the full name. DoubleType is used to define numeric values with decimals in your data. The Pipeline tool helps organize different steps in a machine learning workflow. Many classification models like Logistic Regression, Decision Tree, Random Forest, and Naive Bayes are loaded so you can train and test different algorithms. Evaluators are included to check how good your models are at making predictions. Finally, tools for preparing data (like scaling numbers, combining features, and handling missing values) and indexing text labels are also imported to help with data preprocessing before training your model.
from pyspark.sql import SparkSession
from pyspark.sql import Window, Row
from pyspark.sql.functions import (
col, when, count, lit, expr, rand, hour, dayofweek, mean,
stddev, round, log1p, coalesce, unix_timestamp
)
import pyspark.sql.functions as F
from pyspark.sql.types import DoubleType
from pyspark.ml import Pipeline
from pyspark.ml.classification import (
LogisticRegression, DecisionTreeClassifier, RandomForestClassifier,
GBTClassifier, NaiveBayes, LinearSVC, MultilayerPerceptronClassifier
)
from pyspark.ml.evaluation import BinaryClassificationEvaluator, MulticlassClassificationEvaluator
from pyspark.ml.feature import (
VectorAssembler, MinMaxScaler, StandardScaler, Imputer
)
from pyspark.ml.linalg import VectorUDT, Vectors
from pyspark.sql.functions import udf
from pyspark.ml.feature import StringIndexer
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession
This code imports key libraries for data analysis and visualization.pandas is used to handle and manipulate tabular data (DataFrames).plotly.express allows for quick and simple creation of interactive charts.random is a Python module used here to generate random values if needed.plotly.graph_objects offers more advanced a nd customizable plot-building tools.Together, these tools help create interactive data visualizations from structured datasets.
import pandas as pd
import plotly.express as px
import random
import plotly.graph_objects as go
This below code create and configure a Spark session in PySpark. The SparkSession.builder initializes the session, with the .appName("AML_25") setting the application name. The .master("yarn") specifies using YARN for resource management, while .config("spark.hadoop.fs.defaultFS", "hdfs://dsm-master:9000") configures Spark to connect to the HDFS at the given URI. Finally, .getOrCreate() ensures the session is either created or retrieved if it already exists.
spark = SparkSession.builder \
.appName("AML_25") \
.master("yarn") \
.config("spark.hadoop.fs.defaultFS", "hdfs://dsm-master:9000") \
.getOrCreate()
The code reads a CSV file called "SAML-D.csv" into a Spark DataFrame from the cluster. It automatically uses the first row as column names and tries to figure out the data types of each column. Then, it shows the first 5 rows of the DataFrame to give a preview of the data. Finally, it prints out the list of column names in the DataFrame.
df = spark.read.csv("hdfs:///user/ataga001/SAML-D.csv", header=True, inferSchema=True)
df.show(5)
print("Columns in DataFrame:", df.columns)
The output shows the first 5 rows of the DataFrame with columns describing transaction details, such as time, date, sender and receiver accounts, amount, currencies, bank locations, payment types, and laundering information. It also lists the column names of the DataFrame, including transaction time, sender and receiver details, and laundering flags. The dataset contains details about different types of payments, such as cash deposits, cross-border transfers, and cheques. The column "Is_laundering" marks whether the transaction is suspected of money laundering.
This code snippet is used to determine and display the dimensions of a DataFrame. First, it calculates the number of columns by using len(df.columns), which counts all the column names present in the DataFrame. Then, it computes the number of rows using df.count(), which in PySpark returns the total number of records (in Pandas, this would typically return non-null counts per column). Finally, both values are printed out using formatted strings, providing a quick summary of the DataFrame’s structure.
num_cols = len(df.columns)
num_rows = df.count()
print(f"Number of columns in DataFrame: {num_cols}")
print(f"Number of rows in DataFrame: {num_rows}")
df.select("Amount", "Is_laundering").explain()
This output shows the DataFrame has 12 columns and around 9.5 million rows.The "Physical Plan" comes from Spark and shows how the data is being read.It’s scanning a CSV file stored in HDFS at the given path. Only two columns, Amount and Is_laundering, are being read with their data types (double and int).
The below code converts the DataFrame df into an RDD to work with lower-level Spark operations. It extracts only the Is_laundering column to create a new RDD. Then, it counts how many values are 1 (laundering) and 0 (non-laundering). Finally, it prints both counts and lists all column names in the original DataFrame.
rdd = df.rdd
laundering_rdd = rdd.map(lambda x: x['Is_laundering'])
laundering_count = laundering_rdd.filter(lambda x: x == 1).count()
non_laundering_count = laundering_rdd.filter(lambda x: x == 0).count()
print(f"Number of laundering transactions: {laundering_count}")
print(f"Number of non-laundering transactions: {non_laundering_count}")
print("Column Names:", df.columns)
This dataset records individual financial transfers, detailing who sent money to whom, when, and how much. It also specifies the currencies used and the locations of the sending and receiving banks. Crucially, it includes flags indicating if a transaction is suspected of money laundering and, if so, the type of laundering involved.
This below code snippet analyzes the different types of payments within a Spark DataFrame. It first groups the data by "Payment_type" and counts the occurrences of each type, then orders these counts in descending order. Finally, it uses the Plotly library to create an interactive bar chart visualizing the number of transactions for each payment type, making it easy to compare their frequencies.
payment_counts = df.groupBy("Payment_type").count().orderBy("count", ascending=False)
payment_counts_pd = payment_counts.toPandas()
fig = px.bar(payment_counts_pd,
x="Payment_type",
y="count",
color="Payment_type",
title="Transaction Counts per Payment Type",
labels={"count": "Number of Transactions", "Payment_type": "Payment Type"},
template="plotly_white")
fig.update_layout(xaxis_tickangle=-45)
fig.show()
The first two lines of code filter the DataFrame df to count transactions flagged as either laundering (Is_laundering == 1) or not (Is_laundering == 0). The subsequent two print statements then display these counts to provide a direct comparison. The remaining code uses Plotly to create a histogram visualizing this same information, with separate bars and colors for laundering and non-laundering transactions, making the comparison visually clear. The x-axis label is "Is Laundering" (0 or 1), and the y-axis represents the "Count" of transactions.
laundering_count = df.filter(df.Is_laundering == 1).count()
non_laundering_count = df.filter(df.Is_laundering == 0).count()
print(f"Number of laundering transactions: {laundering_count}")
print(f"Number of non-laundering transactions: {non_laundering_count}")
is_laundering_counts = df.groupBy("Is_laundering").count()
is_laundering_pd = is_laundering_counts.toPandas()
fig = px.histogram(is_laundering_pd,
x="Is_laundering",
y="count",
title="Laundering vs. Non-Laundering (Plotly)",
labels={"Is_laundering": "Is Laundering", "count": "Count"},
color="Is_laundering",
color_discrete_sequence=px.colors.qualitative.Set2)
fig.update_layout(bargap=0.3)
fig.show()
These figures reveal a significant imbalance in the dataset. The vast majority of transactions (9,494,979) are classified as non-laundering, while a much smaller number (9,873) are flagged as potential money laundering incidents.
The first part of the below code, using an RDD (Resilient Distributed Dataset), calculates the total transaction amount for each sender account. It maps each row to extract the sender's account and the transaction amount, filters out any transactions with missing amounts, and then aggregates the amounts by sender. Finally, it identifies and prints the top 5 sender accounts with the highest total transaction values.The subsequent part of the code analyzes payment types using a Spark DataFrame. It groups the DataFrame by "Payment_type" and counts the occurrences of each type. This count data is then converted to a Pandas DataFrame and visualized as a pie chart using Plotly, showing the proportion of each payment type within the dataset.
sender_amount_rdd = rdd.map(lambda row: (row['Sender_account'], row['Amount'])) \
.filter(lambda x: x[1] is not None)
sender_total_amounts = sender_amount_rdd.reduceByKey(lambda x, y: x + y)
top_5_senders = sender_total_amounts.takeOrdered(5, key=lambda x: -x[1])
print("Top 5 Sender Accounts by Total Transaction Amount:")
for sender, total_amount in top_5_senders:
print(f"Sender Account: {sender} | Total Amount: {total_amount:.2f}")
These are the top five sender accounts based on the total value of their transactions in the dataset. Account number 1534642148 has the highest total, with over 15 million in transaction value. The other listed accounts follow, each having transacted amounts exceeding 12 million. This information highlights the accounts with the most significant financial activity within the recorded period.
payment_type_counts = df.groupBy("Payment_type").count()
payment_type_pd = payment_type_counts.toPandas()
fig = px.pie(payment_type_pd,
names="Payment_type",
values="count",
title="Distribution of Transactions per Payment Type")
fig.show()
This Python code iterates through each column in your PySpark DataFrame, named df. For every column, it counts the number of missing values, considering both actual null values and empty string representations as missing.
print("\nMissing values in each column (PySpark):")
for column in df.columns:
missing_count = df.filter(col(column).isNull() | (col(column) == "")).count()
print(f"Column {column} has {missing_count} missing values")
The output shows that after checking for missing values in PySpark DataFrame, every column has zero missing entries. This indicates a clean dataset in terms of completeness for all the listed features, including time, date, account details, transaction amounts, currencies, bank locations, payment types, and the laundering flags.
The below code makes the transaction data richer by creating new information. It takes the time and date and pulls out the hour and day. It also counts how many times each sender made a transaction and spots when money moved between different countries. Plus, it changes the transaction amounts slightly using a math trick and figures out the average amount each sender usually sends. Lastly, it looks at how often each sender makes transactions compared to all their transactions, showing a few examples of this new information. These engineered features can improve the performance of machine learning models, especially for tasks like fraud detection or risk assessment.
df = df.withColumn("Hour", hour(col("Time")))
df = df.withColumn("Day_of_Week", dayofweek(col("Date")))
window_spec = Window.partitionBy("Sender_account")
df = df.withColumn("Sender_Txn_Count", count(lit(1)).over(window_spec))
df = df.withColumn("Is_Cross_Border", when(col("Sender_bank_location") != col("Receiver_bank_location"), 1).otherwise(0))
df = df.withColumn("Log_Amount", log1p(col("Amount")))
df = df.withColumn("Avg_Amount", mean("Amount").over(window_spec))
df = df.withColumn("Txn_Frequency", col("Sender_Txn_Count") / count("*").over(window_spec))
df.select("Hour", "Day_of_Week", "Sender_Txn_Count", "Is_Cross_Border", "Log_Amount", "Avg_Amount", "Txn_Frequency").show(5)
This table displays the first five transactions with the newly created features. "Hour" indicates the transaction time. "Day_of_Week" specifies the day the transaction occurred. "Sender_Txn_Count" shows the total transactions by that sender. "Is_Cross_Border" flags international transfers. "Log_Amount" is a transformed transaction amount. "Avg_Amount" is the sender's average transaction size. "Txn_Frequency" represents this transaction's proportion of the sender's total activity (initially showing as 1.0, suggesting these might be the first recorded transactions for these senders within the window).
Why Winsorizing
This reduces the disproportionate influence of outliers on statistical measures and model training, leading to more strong results. It aims to balance the need for outlier mitigation with the desire to retain as much of the original data distribution as possible.
This code identifies and handles potential outliers in the "Amount" column of your DataFrame. It calculates the first quartile (Q1) and the third quartile (Q3) to determine the interquartile range (IQR). Using the IQR, it establishes lower and upper bounds. Any transaction amount falling below the lower bound is capped at that value, and any amount exceeding the upper bound is capped at the upper bound. Finally, it replaces the original "Amount" column with this capped version and displays descriptive statistics of the adjusted amounts(Reference 3).
quantiles = df.approxQuantile("Amount", [0.25, 0.75], 0.05)
Q1, Q3 = quantiles[0], quantiles[1]
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR
df = df.withColumn("Amount_Capped",
when(col("Amount") < lower_bound, lower_bound)
.when(col("Amount") > upper_bound, upper_bound)
.otherwise(col("Amount")))
df = df.drop("Amount").withColumnRenamed("Amount_Capped", "Amount")
df.select("Amount").describe().show()
This output provides summary statistics for the "Amount" column after the outlier capping process. It shows that there are 9,504,852 transaction records. The average transaction amount is approximately 7343.75, with a standard deviation of about 6133.70, indicating the spread of the amounts. The minimum transaction amount is 3.73, and the maximum amount after capping is 22783.95. These statistics give an overview of the central tendency and distribution of the transaction amounts after handling extreme values.
This code prepares the minority class (laundering transactions) for machine learning. It first filters the DataFrame to isolate these instances. Then, it selects specific columns ("Hour", "Day_of_Week", etc.) considered relevant features for prediction. The VectorAssembler combines these selected features into a single vector column named "features". This "features" column is a standard input format required by many Spark MLlib machine learning algorithms. By assembling the features, we make the data suitable for training a model to detect money laundering.
minority_df = df.filter(col("Is_laundering") == 1)
majority_df = df.filter(col("Is_laundering") == 0)
feature_cols = ["Hour", "Day_of_Week", "Sender_Txn_Count", "Is_Cross_Border", "Log_Amount", "Avg_Amount", "Txn_Frequency"]
assembler = VectorAssembler(inputCols=feature_cols, outputCol="features")
minority_df = assembler.transform(minority_df)
This Python function, generate_synthetic_row, takes a row from a PySpark DataFrame as input and generates a slightly modified, synthetic version of it. It extracts the "features" vector and adds a small amount of random noise to each feature value. The function then constructs a new row with this perturbed "features" vector and sets the "Is_laundering" label to 1, indicating a synthetic laundering instance. It also carries over the original values for other columns, ensuring the synthetic data retains context. This method is called synthetic data generation or, in the context of addressing class imbalance, oversampling using synthetic data augmentation. A specific technique often used for this purpose is SMOTE (Synthetic Minority Over-sampling Technique), although the provided code implements a simpler form of synthetic generation (Reference -2).
$$ f_i^{\text{synthetic}} = f_i + (\text{random factor}) \times f_i $$The synthetic row is generated by adjusting the original feature vector using a random factor between -0.2 and 0.2. Each feature in the vector is modified by multiplying it by a random factor and adding the result to the original value.
def generate_synthetic_row(row, k=10):
feature_vector = row["features"].toArray()
synthetic_vector = feature_vector + random.uniform(-0.2, 0.2) * feature_vector
return Row(
features=Vectors.dense(synthetic_vector),
Is_laundering=1,
Hour=float(synthetic_vector[0]),
Day_of_Week=float(synthetic_vector[1]),
Sender_Txn_Count=float(synthetic_vector[2]),
Is_Cross_Border=float(synthetic_vector[3]),
Log_Amount=float(synthetic_vector[4]),
Avg_Amount=float(synthetic_vector[5]),
Txn_Frequency=float(synthetic_vector[6]),
Time=row["Time"],
Date=row["Date"],
Sender_account=row["Sender_account"],
Receiver_account=row["Receiver_account"],
Payment_currency=row["Payment_currency"],
Received_currency=row["Received_currency"],
Sender_bank_location=row["Sender_bank_location"],
Receiver_bank_location=row["Receiver_bank_location"],
Payment_type=row["Payment_type"],
Laundering_type=row["Laundering_type"],
Amount=row["Amount"]
)
This below code addresses class imbalance by oversampling the minority class (laundering transactions). It first converts the minority DataFrame to an RDD. Then, it applies the generate_synthetic_row function to each minority instance, creating 10 synthetic examples for each original one using a flatMap operation. These synthetic rows are then converted back into a DataFrame. Finally, this synthetic DataFrame is combined with the original majority and minority DataFrames to create a balanced dataset, and the counts of laundering and non-laundering transactions in the balanced dataset are displayed.
minority_rdd = minority_df.rdd
synthetic_rdd = minority_rdd.flatMap(lambda row: [generate_synthetic_row(row) for _ in range(10)])
synthetic_df = spark.createDataFrame(synthetic_rdd)
balanced_df = majority_df.union(minority_df.select(majority_df.columns)).union(synthetic_df.select(majority_df.columns))
balanced_df.groupBy("Is_laundering").count().show()
This output shows the distribution of the target variable, "Is_laundering," in the balanced dataset after the oversampling technique was applied. The count for non-laundering transactions (0) remains at 9,494,979, representing the original majority class. The count for laundering transactions (1) has increased significantly to 108,603 due to the generated synthetic samples, indicating a more balanced class distribution compared to the original data.
The data is divided into two parts: one for training the model and the other for testing it. The split ensures that the data is randomly divided but in a way that can be repeated exactly the same every time. To improve performance, both the training and testing data are stored in memory. This makes future operations faster since the data doesn't need to be reloaded from disk each time it's accessed.
train_data, test_data = balanced_df.randomSplit([0.8, 0.2], seed=42)
train_data.cache()
test_data.cache()
The DataFrame contains transaction details like time, date, sender/receiver account information, and currencies involved. It also includes features such as payment type, laundering status, transaction count, and cross-border indication. Numeric columns like transaction amount, frequency, and averages are included for analysis. This structured data is used for detecting patterns like money laundering.
The below pyspark code calculates the number of rows in both the training and test datasets. It then computes the percentage of data in the training and test sets relative to the total dataset. After that, it prints the sizes of the training, test, and total datasets. Finally, it groups the test data by the "Is_laundering" column and shows the count of each category (laundering or not).
train_count = train_data.count()
test_count = test_data.count()
total_count = train_count + test_count
print(f"Training set size: {train_count} ({train_count/total_count*100:.1f}%)")
print(f"Test set size: {test_count} ({test_count/total_count*100:.1f}%)")
print(f"Total size: {total_count}")
test_data.groupBy("Is_laundering").count().show()
The training set consists of 7,684,886 records, which is 80.0% of the total data, while the test set has 1,918,696 records, or 20.0%. The total dataset size is 9,603,582 records. In the test set, there are 1,897,092 records without laundering (Is_laundering = 0) and 21,604 records with laundering (Is_laundering = 1).
The below code calculates the number of fraudulent (Is_laundering = 1) and non-fraudulent (Is_laundering = 0) transactions in the training dataset. It then computes a weight for each class (fraud and non-fraud) to address any class imbalance. The weights are inversely proportional to the number of samples in each class, making the less frequent class (fraud) more important in model training.
fraud_count = train_data.filter(col("Is_laundering") == 1).count()
non_fraud_count = train_data.filter(col("Is_laundering") == 0).count()
total_count = fraud_count + non_fraud_count
fraud_weight = total_count / (2.0 * fraud_count)
non_fraud_weight = total_count / (2.0 * non_fraud_count)
train_data = train_data.withColumn(
"weight",
when(col("Is_laundering") == 1, fraud_weight).otherwise(non_fraud_weight)
)
test_data = test_data.withColumn(
"weight",
when(col("Is_laundering") == 1, fraud_weight).otherwise(non_fraud_weight)
)
The weight column is added to both the training and test datasets, where fraudulent transactions get the fraud weight and non-fraudulent ones get the non-fraud weight. This ensures the model places more importance on detecting fraud during training and evaluation.
Classification models predict if something belongs to a category, like identifying fraudulent transactions. They analyze data features, such as transaction amounts or locations. Trained models, like Decision Trees, learn patterns for accurate predictions. Simpler models, like Naive Bayes, often struggle to separate fraud from non-fraud. Comparing performance metrics helps choose the best model
In the code, PySpark is used for distributed data processing. It utilizes withColumn to add prediction columns to the dataset and a UDF to create custom prediction vectors. Evaluators like MulticlassClassificationEvaluator and BinaryClassificationEvaluator are used to compute metrics such as accuracy, precision, recall, and ROC AUC. The groupBy function helps generate a confusion matrix to manually calculate precision and recall for fraud detection. Lastly, unpersist is called to remove the cached DataFrame, freeing up memory. This approach ensures efficient processing of large datasets. We put this model so we can compare the trained models with this to see how they perform better when data is trained
vec_udf = udf(lambda: Vectors.dense([1.0, 0.0]), VectorUDT())
baseline_predictions = test_data.withColumn("prediction", lit(1.0)).withColumn("rawPrediction", vec_udf())
baseline_predictions.cache()
evaluator_accuracy = MulticlassClassificationEvaluator(labelCol="Is_laundering", predictionCol="prediction", metricName="accuracy")
evaluator_precision = MulticlassClassificationEvaluator(labelCol="Is_laundering", predictionCol="prediction", metricName="weightedPrecision")
evaluator_recall = MulticlassClassificationEvaluator(labelCol="Is_laundering", predictionCol="prediction", metricName="weightedRecall")
evaluator_f1 = MulticlassClassificationEvaluator(labelCol="Is_laundering", predictionCol="prediction", metricName="f1")
evaluator_roc = BinaryClassificationEvaluator(labelCol="Is_laundering", rawPredictionCol="rawPrediction", metricName="areaUnderROC")
confusion_matrix = baseline_predictions.groupBy("Is_laundering", "prediction").count().collect()
tp = fp = fn = tn = 0
for row in confusion_matrix:
label, pred, count = row["Is_laundering"], row["prediction"], row["count"]
if label == 1 and pred == 1: tp = count
elif label == 0 and pred == 1: fp = count
elif label == 1 and pred == 0: fn = count
elif label == 0 and pred == 0: tn = count
precision_pos = tp / (tp + fp) if (tp + fp) > 0 else 0.0
recall_pos = tp / (tp + fn) if (tp + fn) > 0 else 0.0
baseline_metrics = {
"Accuracy": evaluator_accuracy.evaluate(baseline_predictions),
"Precision (Weighted)": evaluator_precision.evaluate(baseline_predictions),
"Recall (Weighted)": evaluator_recall.evaluate(baseline_predictions),
"F1 Score (Weighted)": evaluator_f1.evaluate(baseline_predictions),
"ROC AUC": evaluator_roc.evaluate(baseline_predictions),
"Precision (Laundered - label 1)": precision_pos,
"Recall (Laundered - label 1)": recall_pos,
"True Negatives": tn
}
print("Untrained Baseline (All Fraud - label 1) Metrics:")
for metric, value in baseline_metrics.items():
print(f"{metric}: {value:.3f}")
baseline_predictions.unpersist()
The baseline model, which predicts all transactions as fraudulent (label 1), has poor performance, indicated by low accuracy (0.011) and precision (0.000). The recall for fraud cases (label 1) is 1.000, meaning it correctly identifies all fraud cases, but precision is extremely low, suggesting a high number of false positives. The F1 score and weighted precision are also near zero, reflecting poor overall model performance. The confusion matrix shows no true negatives, indicating the model fails to correctly classify non-fraudulent transactions. The dataset includes various features related to transactions, including time, account details, and amounts, alongside the predictions and calculated weights.
Logistic regression is a statistical method used for binary classification, where the output is modeled as the probability of a particular class (e.g., fraud detection). The formula for logistic regression is:
$$ P(y = 1|X) = \frac{1}{1 + e^{-(\beta_0 + \beta_1 X_1 + \beta_2 X_2 + \dots + \beta_n X_n)}} $$Where $ P(y = 1|X) $ is the probability that the observation belongs to class 1 (fraud), $ X_1, X_2, \dots, X_n $ are the feature values, and $ \beta_0, \beta_1, \dots, \beta_n $ are the coefficients (weights) of the model. The output is transformed using the logistic (sigmoid) function to ensure the result lies between 0 and 1.
This code defines two groups of features: categorical and numeric. Categorical features like "Payment_currency" and "Laundering_type" are transformed into indexed columns for machine learning. Numeric features like "Sender_Txn_Count" and "Amount" are directly used for modeling. The assembler_inputs combines both sets of features into a list that will be used to prepare data for model training.
categorical_columns = [
"Payment_currency", "Received_currency", "Sender_bank_location",
"Receiver_bank_location", "Payment_type", "Laundering_type"
]
numeric_features = [
"Sender_Txn_Count", "Is_Cross_Border", "Log_Amount", "Amount", "Hour", "Day_of_Week",
"Avg_Amount", "Txn_Frequency"
]
assembler_inputs = numeric_features + [f"{col_name}_indexed" for col_name in categorical_columns]
In PySpark, this code fills missing (null) values in the columns of the train_data and test_data DataFrames. For numeric columns (such as "Sender_Txn_Count" or "Amount"), it uses the fillna() function to replace any null values with 0. For categorical columns (such as "Payment_currency" or "Laundering_type"), it replaces null values with the string "Unknown". This ensures that the datasets are complete and ready for machine learning models without any missing values.
for col_name in numeric_features:
train_data = train_data.fillna({col_name: 0})
test_data = test_data.fillna({col_name: 0})
for col_name in categorical_columns:
train_data = train_data.fillna({col_name: "Unknown"})
test_data = test_data.fillna({col_name: "Unknown"})
In this code, StringIndexer is used to convert categorical columns into numeric indices, storing the result in new columns with the suffix _indexed. VectorAssembler combines both the numeric and indexed categorical columns into a single vector column called "features" for model input. StandardScaler then standardizes the "features" column to have a mean of 0 and a standard deviation of 1, improving the performance of models that are sensitive to feature scaling. This preparation is essential for training machine learning models efficiently in PySpark.
indexers = [StringIndexer(inputCol=col_name, outputCol=f"{col_name}_indexed", handleInvalid="keep") for col_name in categorical_columns]
assembler = VectorAssembler(inputCols=assembler_inputs, outputCol="features", handleInvalid="keep")
scaler = StandardScaler(inputCol="features", outputCol="scaled_features", withStd=True, withMean=True)
The code trains a Logistic Regression model on preprocessed features, including both numeric and categorical data, using a Pipeline. The pipeline stages include indexing categorical columns, assembling features, scaling them, and fitting the model. Once the model is trained, predictions are made on both the training and test datasets. The evaluate_model function computes key metrics like accuracy, precision, recall, F1 score, and ROC AUC, by calculating true positives (TP), false positives (FP), false negatives (FN), and true negatives (TN) from the predictions. BinaryClassificationEvaluator and MulticlassClassificationEvaluator are used to evaluate these metrics. The metrics are displayed, printed, and stored in a Pandas DataFrame for further inspection. This entire process is executed using PySpark on a distributed system HDFS, ensuring scalable data processing and model training.
lr_model = LogisticRegression(featuresCol="scaled_features", labelCol="Is_laundering", weightCol="weight", regParam=0.01, elasticNetParam=0.5, maxIter=100, tol=1e-6)
pipeline = Pipeline(stages=indexers + [assembler, scaler, lr_model])
model = pipeline.fit(train_data)
train_assembled = model.transform(train_data)
test_assembled = model.transform(test_data)
predictions = model.transform(test_data)
def evaluate_model(predictions):
tp = predictions.filter((col("Is_laundering") == 1) & (col("prediction") == 1)).count()
fp = predictions.filter((col("Is_laundering") == 0) & (col("prediction") == 1)).count()
fn = predictions.filter((col("Is_laundering") == 1) & (col("prediction") == 0)).count()
tn = predictions.filter((col("Is_laundering") == 0) & (col("prediction") == 0)).count()
binary_evaluator = BinaryClassificationEvaluator(labelCol="Is_laundering", rawPredictionCol="rawPrediction", metricName="areaUnderROC")
multi_evaluator = MulticlassClassificationEvaluator(labelCol="Is_laundering", predictionCol="prediction")
metrics = {
"Model": "Logistic Regression",
"Accuracy": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "accuracy"}),
"Precision": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "weightedPrecision"}),
"Recall": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "weightedRecall"}),
"F1 Score": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "f1"}),
"ROC AUC": binary_evaluator.evaluate(predictions),
"Laundered Precision": tp / (tp + fp) if (tp + fp) > 0 else 0,
"Laundered Recall": tp / (tp + fn) if (tp + fn) > 0 else 0,
"True Negatives": tn
}
return metrics
metrics = evaluate_model(predictions)
print("\nLogistic Regression Results:")
print("{:<20} {:<10}".format('Metric', 'Value'))
print("-" * 30)
for metric, value in metrics.items():
print("{:<20} {:<10.3f}".format(metric, value) if isinstance(value, (int, float)) else "{:<20} {:<10}".format(metric, value))
lr_results_df = pd.DataFrame(list(metrics.items()), columns=["Metric", "Value"])
print("\nStored Logistic Regression Results in DataFrame:")
display(lr_results_df)
The Logistic Regression model achieved high performance with an accuracy of 0.994, indicating it predicted most instances correctly. Precision of 0.996 shows its effectiveness at identifying fraudulent transactions with minimal false positives, while recall of 0.994 reflects its ability to identify most actual fraud cases with low false negatives. The F1 score of 0.995 demonstrates balanced precision and recall. A perfect ROC AUC of 1.000 indicates excellent class discrimination. Laundered precision is 0.653, showing good performance in identifying fraudulent transactions relative to positive predictions, while laundered recall is perfect at 1. True negatives totaled 1,885,591. These results, processed using PySpark on HDFS, are stored in a DataFrame for further analysis.
This below code filters predictions to select only fraudulent transactions (Is_laundering == 1) and selects relevant columns. It limits the result to 5 rows for a quick sample. A UDF (extract_prob) is defined to extract the fraud probability from the probability column and is applied to create a new fraud_probability column. The code then selects and displays specific features, including fraud_probability and prediction, without truncating the output. PySpark is used to handle the data efficiently, which could be stored in HDFS. This allows scalable processing on large datasets.
sample_fraud_prediction = predictions.filter(col("Is_laundering") == 1).select(
"Sender_account", "Receiver_account", "Amount", "Payment_currency", "Received_currency",
"Sender_bank_location", "Receiver_bank_location", "Payment_type", "Laundering_type",
"Hour", "Day_of_Week", "Sender_Txn_Count", "Is_Cross_Border", "Log_Amount",
"Avg_Amount", "Txn_Frequency", "Is_laundering", "probability", "prediction"
).limit(5)
def extract_prob(prob):
return float(prob[1])
extract_prob_udf = udf(extract_prob, DoubleType())
sample_fraud_prediction = sample_fraud_prediction.withColumn("fraud_probability", extract_prob_udf(col("probability")))
sample_fraud_prediction.select(
"Sender_account", "Receiver_account", "Amount", "Payment_type", "Is_Cross_Border",
"Sender_Txn_Count", "Avg_Amount", "Txn_Frequency", "Is_laundering",
"fraud_probability", "prediction"
).show(truncate=False)
The output suggests that the model has identified potentially fraudulent transactions, with all predicted fraud cases having a prediction value of 1. Analyzing the fraud_probability values, we see that the model is quite confident in its predictions, with probabilities ranging from 0.80 to 0.99. These high probabilities imply that the model is making accurate predictions for transactions it deems fraudulent. Since the predicted values match the actual values for all five transactions, all five predictions are correct.
The below code creates a bar chart to visualize the performance metrics of the Logistic Regression model. It first filters out non-numeric values from the lr_results_df DataFrame and then generates random colors for each metric using the HSL color model. Using Plotly's px.bar, it creates a bar chart where each metric (e.g., Accuracy, Precision, Recall) is displayed with its corresponding value, formatting the text outside the bars and ensuring the Y-axis is scaled between 0 and 1.
lr_results_df = pd.DataFrame(list(metrics.items()), columns=["Metric", "Value"])
plot_df = lr_results_df[lr_results_df["Value"].apply(lambda x: isinstance(x, (int, float)))]
colors = ["hsl({}, 70%, 50%)".format(random.randint(0, 360)) for _ in range(len(plot_df))]
fig = px.bar(plot_df, x="Metric", y="Value", text="Value", color="Metric", color_discrete_sequence=colors)
fig.update_traces(texttemplate='%{text:.3f}', textposition='outside')
fig.update_layout(uniformtext_minsize=8, uniformtext_mode='hide', yaxis_range=[0, 1.05])
fig.show()
The Decision Tree classifier builds a model by recursively splitting the dataset into subsets based on feature values to maximize class purity. It uses a splitting criterion like Gini impurity or entropy. In this code, the Decision Tree model uses "gini" impurity, with maximum depth of 10 and minimum instances per node set to 50. Key metrics evaluated include accuracy, precision, recall, F1 score, ROC AUC, laundered precision, laundered recall, and true negatives. For classification, the formula for splitting a node is:
The Gini Index is calculated as:
$$ {Gini Index} = 1 - \sum_{i=1}^{k} p_i^2 $$where $ p_i $ is the probability of a class in a given node. The Decision Tree aims to minimize this index at each split, selecting features that best separate the classes.
In below code, categorical columns are indexed using StringIndexer, converting them into numerical values, which is essential for machine learning models in PySpark. The VectorAssembler combines both categorical and numeric features into a single feature vector, while the StandardScaler normalizes these features for better model performance. Missing values in both train and test datasets are handled by filling categorical columns with "Unknown" and numeric columns with 0. This data processing pipeline is designed to run on PySpark, enabling efficient distributed processing on HDFS for large-scale datasets.
categorical_columns = ["Payment_currency", "Received_currency", "Sender_bank_location", "Receiver_bank_location", "Payment_type"]
numeric_features = ["Sender_Txn_Count", "Is_Cross_Border", "Log_Amount", "Amount", "Hour", "Day_of_Week", "Avg_Amount", "Txn_Frequency"]
assembler_inputs = numeric_features + [f"{col_name}_indexed" for col_name in categorical_columns]
for col_name in numeric_features:
train_data = train_data.fillna({col_name: 0})
test_data = test_data.fillna({col_name: 0})
for col_name in categorical_columns:
train_data = train_data.fillna({col_name: "Unknown"})
test_data = test_data.fillna({col_name: "Unknown"})
indexers = [StringIndexer(inputCol=col_name, outputCol=f"{col_name}_indexed", handleInvalid="keep") for col_name in categorical_columns]
assembler = VectorAssembler(inputCols=assembler_inputs, outputCol="features", handleInvalid="keep")
scaler = StandardScaler(inputCol="features", outputCol="scaled_features", withStd=True, withMean=True)
This code builds and evaluates a Decision Tree model using PySpark on a dataset stored in HDFS. The pipeline first processes categorical columns with StringIndexer, assembles features with VectorAssembler, scales the features using StandardScaler, and trains the DecisionTreeClassifier on the train_data. The model's performance is evaluated using metrics like accuracy, precision, recall, F1 score, and ROC AUC. Finally, the evaluation results are printed and stored in a Pandas DataFrame for further analysis.
dt_model = DecisionTreeClassifier(featuresCol="scaled_features", labelCol="Is_laundering", maxDepth=10, minInstancesPerNode=50, impurity="gini", seed=42)
pipeline = Pipeline(stages=indexers + [assembler, scaler, dt_model])
model = pipeline.fit(train_data)
predictions = model.transform(test_data)
def evaluate_model(predictions):
tp = predictions.filter((col("Is_laundering") == 1) & (col("prediction") == 1)).count()
fp = predictions.filter((col("Is_laundering") == 0) & (col("prediction") == 1)).count()
fn = predictions.filter((col("Is_laundering") == 1) & (col("prediction") == 0)).count()
tn = predictions.filter((col("Is_laundering") == 0) & (col("prediction") == 0)).count()
binary_evaluator = BinaryClassificationEvaluator(labelCol="Is_laundering", rawPredictionCol="rawPrediction", metricName="areaUnderROC")
multi_evaluator = MulticlassClassificationEvaluator(labelCol="Is_laundering", predictionCol="prediction")
metrics = {
"Model": "Decision Tree",
"Accuracy": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "accuracy"}),
"Precision": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "weightedPrecision"}),
"Recall": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "weightedRecall"}),
"F1 Score": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "f1"}),
"ROC AUC": binary_evaluator.evaluate(predictions),
"Laundered Precision": tp / (tp + fp) if (tp + fp) > 0 else 0,
"Laundered Recall": tp / (tp + fn) if (tp + fn) > 0 else 0,
"True Negatives": tn
}
return metrics
metrics = evaluate_model(predictions)
print("\nDecision Tree Results:")
print("{:<20} {:<10}".format('Metric', 'Value'))
print("-" * 30)
for metric, value in metrics.items():
print("{:<20} {:<10.3f}".format(metric, value) if isinstance(value, (int, float)) else "{:<20} {:<10}".format(metric, value))
dt_results_df = pd.DataFrame(list(metrics.items()), columns=["Metric", "Value"])
print("\nStored Decision Tree Results in DataFrame:")
display(dt_results_df)
The Decision Tree model's performance metrics indicate very high accuracy, precision, and recall, all above 0.998, suggesting it performs well in identifying both positive and negative cases. The ROC AUC score of 0.96 shows a strong ability to distinguish between the classes. The "Laundered Precision" and "Laundered Recall" values, at 0.99 and 0.91, respectively, show that the model effectively identifies potential fraudulent transactions. The count of "True Negatives" is very high, with over 1.8 million correctly classified non-laundered transactions. These results are displayed and stored in a DataFrame for further analysis.
This code selects a random sample of 3 predictions from the model results using orderBy(rand()).limit(3). It then applies a user-defined function (extract_prob_udf) to extract the fraud probability from the "probability" column. The selected columns, including fraud probability and the prediction, are displayed with the .show(truncate=False) method for full visibility in the output. This operation is performed on the distributed data stored in HDFS using PySpark to handle large-scale processing and transformation.
sample_combined = predictions.orderBy(rand()).limit(3)
sample_combined = sample_combined.withColumn("fraud_probability", extract_prob_udf(col("probability")))
sample_combined.select(
"Sender_account", "Receiver_account", "Amount", "Payment_type", "Is_Cross_Border",
"Sender_Txn_Count", "Avg_Amount", "Txn_Frequency", "Is_laundering",
"fraud_probability", "prediction"
).show(truncate=False)
The displayed output presents a sample of 3 transactions, each showing the sender and receiver account, transaction amount, payment type, cross-border status, transaction count, average amount, frequency, and laundering label. The "fraud_probability" column indicates the likelihood of fraud detected by the model, while the "prediction" column shows the model’s classification (0 for non-fraud, 1 for fraud). .Since the predicted value ("0.0") matches the actual value (Is_laundering = 0) for all three transactions, all three predictions are correct.
This code creates a bar chart visualizing the evaluation metrics of the Decision Tree model. It first converts the metrics dictionary into a DataFrame, filters out non-numeric values, and assigns random colors for each metric. A bar chart is then generated using Plotly Express, displaying each metric’s value, with the text displayed outside the bars for clarity. The layout is customized for better readability, with a title and y-axis range set for consistency.
dt_results_df = pd.DataFrame(list(metrics.items()), columns=["Metric", "Value"])
plot_df = dt_results_df[dt_results_df["Value"].apply(lambda x: isinstance(x, (int, float)))]
colors = ["hsl({}, 70%, 50%)".format(random.randint(0, 360)) for _ in range(len(plot_df))]
fig = px.bar(plot_df, x="Metric", y="Value", text="Value", color="Metric", color_discrete_sequence=colors)
fig.update_traces(texttemplate='%{text:.3f}', textposition='outside')
fig.update_layout(
uniformtext_minsize=8,
uniformtext_mode='hide',
yaxis_range=[0, 1.05],
title="Decision Tree Evaluation Metrics"
)
fig.show()
Random Forest is an ensemble learning method that builds multiple decision trees and combines their results to improve prediction accuracy and reduce overfitting. It works by randomly selecting subsets of features and data points to train each tree, and then aggregates the predictions (typically by majority vote or averaging) for classification or regression tasks. The formula for a Random Forest model is as follows:
$$ Text{Prediction} = \frac{1}{N} \sum_{i=1}^{N} T_i(x) $$Where: $ T_i(x) $ is the prediction of the $ i-th $ tree for input $ x $ $ N $ is the total number of trees in the Random Forest.
In this process, data from the HDFS (Hadoop Distributed File System) is used for training and testing a Random Forest model in PySpark. The data is first preprocessed by combining relevant features into a feature vector using VectorAssembler. Afterward, a Random Forest model is trained on the distributed dataset using multiple decision trees to classify the target variable (Is_laundering). Once the model is trained, it makes predictions on unseen test data, leveraging Spark’s parallel processing for efficient computation on large-scale datasets.
input_cols = ['Hour', 'Day_of_Week', 'Sender_Txn_Count', 'Is_Cross_Border', 'Log_Amount', 'Amount']
assembler = VectorAssembler(inputCols=input_cols, outputCol="features")
train_assembled = assembler.transform(train_data)
test_assembled = assembler.transform(test_data)
model = RandomForestClassifier(featuresCol="features", labelCol="Is_laundering", numTrees=20)
fitted_model = model.fit(train_assembled)
predictions = fitted_model.transform(test_assembled)
This process evaluates the performance of a Random Forest model in PySpark using HDFS data. It calculates the confusion matrix elements: true positives (TP), false positives (FP), false negatives (FN), and true negatives (TN) from predictions on data stored in HDFS. The model's performance is then measured with common metrics like Accuracy, Precision, Recall, F1 Score, and ROC AUC, along with custom metrics like Laundered Precision and Recall. Finally, the results are displayed using PySpark DataFrames, and the metrics are stored for further analysis or reporting.
def evaluate_model_rf(predictions):
tp = predictions.filter((col("Is_laundering") == 1) & (col("prediction") == 1)).count()
fp = predictions.filter((col("Is_laundering") == 0) & (col("prediction") == 1)).count()
fn = predictions.filter((col("Is_laundering") == 1) & (col("prediction") == 0)).count()
tn = predictions.filter((col("Is_laundering") == 0) & (col("prediction") == 0)).count()
binary_evaluator = BinaryClassificationEvaluator(labelCol="Is_laundering", rawPredictionCol="rawPrediction", metricName="areaUnderROC")
multi_evaluator = MulticlassClassificationEvaluator(labelCol="Is_laundering", predictionCol="prediction")
metrics = {
"Model": "Random Forest",
"Accuracy": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "accuracy"}),
"Precision": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "weightedPrecision"}),
"Recall": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "weightedRecall"}),
"F1 Score": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "f1"}),
"ROC AUC": binary_evaluator.evaluate(predictions),
"Laundered Precision": tp / (tp + fp) if (tp + fp) > 0 else 0,
"Laundered Recall": tp / (tp + fn) if (tp + fn) > 0 else 0,
"True Negatives": tn
}
return metrics
metrics_rf = evaluate_model_rf(predictions)
print("\nRandom Forest Results:")
print("{:<20} {:<10}".format('Metric', 'Value'))
print("-" * 30)
for metric, value in metrics_rf.items():
print("{:<20} {:<10.3f}".format(metric, value) if isinstance(value, (int, float)) else "{:<20} {:<10}".format(metric, value))
rf_results_df = pd.DataFrame(list(metrics_rf.items()), columns=["Metric", "Value"])
print("\nStored Random Forest Results in DataFrame:")
display(rf_results_df)
The Random Forest model's evaluation results show high performance with an accuracy, precision, and recall of around 99%, indicating strong predictive power. The F1 score of 0.989 further suggests a balanced model between precision and recall. However, the ROC AUC of 0.913 indicates room for improvement in distinguishing between classes, while the Laundered Precision is 1, showing no false positives in detecting laundering. The Laundered Recall is 0.22, reflecting the model's limited ability to correctly identify laundering cases.
In PySpark, a random sample of 3 predictions is selected using the orderBy(rand()) method, which shuffles the data. A user-defined function (extract_prob_udf) is applied to extract the fraud probability from the model’s prediction. The selected features such as sender/receiver accounts, transaction details, and model outputs like fraud probability and prediction are then displayed using show(). These operations are performed on data stored in HDFS, utilizing PySpark for distributed processing and manipulation.
sample_combined = predictions.orderBy(rand()).limit(3)
sample_combined = sample_combined.withColumn("fraud_probability", extract_prob_udf(col("probability")))
sample_combined.select(
"Sender_account", "Receiver_account", "Amount", "Payment_type", "Is_Cross_Border",
"Sender_Txn_Count", "Avg_Amount", "Txn_Frequency", "Is_laundering",
"fraud_probability", "prediction"
).show(truncate=False)
This output shows a random sample of 3 transactions with details like sender/receiver accounts, transaction amounts, and payment types. It includes the model's prediction on whether the transaction is laundering (Is_laundering), the fraud probability (fraud_probability), and the predicted outcome (prediction).Since the predicted values match the actual values for all three transactions, all three predictions are correct. These results are obtained using PySpark's distributed processing on HDFS-stored data, highlighting the use of machine learning for fraud detection.
The code creates a bar plot using Plotly to visualize the evaluation metrics of the Random Forest model. It first filters out non-numeric values from the results dataframe and assigns a unique color to each metric using a random color generator. The plot displays the metrics (such as Accuracy, Precision, Recall) on the x-axis and their corresponding values on the y-axis, with the values shown outside the bars. The plot’s layout is customized with a title, specific y-axis range, and adjusted text display settings. Finally, the plot is displayed to provide a clear and interactive visualization of the model's performance.
rf_results_df = pd.DataFrame(list(metrics_rf.items()), columns=["Metric", "Value"])
plot_df_rf = rf_results_df[rf_results_df["Value"].apply(lambda x: isinstance(x, (int, float)))]
colors_rf = ["hsl({}, 70%, 50%)".format(random.randint(0, 360)) for _ in range(len(plot_df_rf))]
fig_rf = px.bar(
plot_df_rf,
x="Metric",
y="Value",
text="Value",
color="Metric",
color_discrete_sequence=colors_rf
)
fig_rf.update_traces(texttemplate='%{text:.3f}', textposition='outside')
fig_rf.update_layout(
uniformtext_minsize=8,
uniformtext_mode='hide',
yaxis_range=[0, 1.05],
title="Random Forest Evaluation Metrics"
)
fig_rf.show()
Gradient Boosting Trees (GBT) is an ensemble learning method where multiple decision trees are combined to form a stronger model. Each new tree attempts to correct the errors made by the previous trees by focusing on residuals or errors from the prior iterations. The trees are trained sequentially, with each tree trying to minimize a given loss function, such as Mean Squared Error (MSE) for regression or cross-entropy for classification. The final prediction is a weighted sum of the predictions from all the trees. This model works well for both regression and classification tasks and can handle various types of data. The main advantage of GBT is its ability to focus on difficult-to-predict samples, improving accuracy. However, it is prone to overfitting if not properly tuned.
$$ f(x) = \sum_{i=1}^{M} T_i(x) $$Where:
$ 𝑓(𝑥)f(x) $ is the final prediction.
$ 𝑀M $ is the number of trees.
$𝑇𝑖(𝑥)T i(x)$ is the prediction of the $i-th$ tree.
The code prepares the dataset by assembling the selected input features ('Hour', 'Day_of_Week', etc.) into a single vector column. It then creates and trains a Gradient Boosted Trees (GBT) model on the training data to predict the "Is_laundering" label. Finally, the trained model is used to make predictions on the test data.
input_cols = ['Hour', 'Day_of_Week', 'Sender_Txn_Count', 'Is_Cross_Border', 'Log_Amount', 'Amount']
assembler = VectorAssembler(inputCols=input_cols, outputCol="features")
train_assembled = assembler.transform(train_data)
test_assembled = assembler.transform(test_data)
model = GBTClassifier(featuresCol="features", labelCol="Is_laundering", maxIter=10)
fitted_model = model.fit(train_assembled)
predictions = fitted_model.transform(test_assembled)
The code evaluates the performance of a Gradient-Boosted Trees (GBT) model in PySpark by calculating several classification metrics such as accuracy, precision, recall, F1 score, and ROC AUC. It computes true positives, false positives, false negatives, and true negatives for more granular metrics like laundered precision and recall. PySpark's BinaryClassificationEvaluator and MulticlassClassificationEvaluator are used to assess model performance. The results are then displayed and stored in a Pandas DataFrame for easy interpretation. The underlying data is likely stored in HDFS, which PySpark processes in a distributed manner.
def evaluate_model_gbt(predictions):
tp = predictions.filter((col("Is_laundering") == 1) & (col("prediction") == 1)).count()
fp = predictions.filter((col("Is_laundering") == 0) & (col("prediction") == 1)).count()
fn = predictions.filter((col("Is_laundering") == 1) & (col("prediction") == 0)).count()
tn = predictions.filter((col("Is_laundering") == 0) & (col("prediction") == 0)).count()
binary_evaluator = BinaryClassificationEvaluator(labelCol="Is_laundering", rawPredictionCol="rawPrediction", metricName="areaUnderROC")
multi_evaluator = MulticlassClassificationEvaluator(labelCol="Is_laundering", predictionCol="prediction")
metrics = {
"Model": "Gradient-Boosted Trees",
"Accuracy": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "accuracy"}),
"Precision": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "weightedPrecision"}),
"Recall": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "weightedRecall"}),
"F1 Score": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "f1"}),
"ROC AUC": binary_evaluator.evaluate(predictions),
"Laundered Precision": tp / (tp + fp) if (tp + fp) > 0 else 0,
"Laundered Recall": tp / (tp + fn) if (tp + fn) > 0 else 0,
"True Negatives": tn
}
return metrics
metrics_gbt = evaluate_model_gbt(predictions)
print("\nGradient-Boosted Trees Results:")
print("{:<20} {:<10}".format('Metric', 'Value'))
print("-" * 30)
for metric, value in metrics_gbt.items():
print("{:<20} {:<10.3f}".format(metric, value) if isinstance(value, (int, float)) else "{:<20} {:<10}".format(metric, value))
gbt_results_df = pd.DataFrame(list(metrics_gbt.items()), columns=["Metric", "Value"])
print("\nStored Gradient-Boosted Trees Results in DataFrame:")
display(gbt_results_df)
The Gradient-Boosted Trees (GBT) model achieved high performance with an accuracy of 0.99, precision of 0.99, and recall of 0.99, demonstrating its ability to correctly identify both positive and negative instances. The F1 score of 0.990 suggests a good balance between precision and recall. The ROC AUC value of 0.91 indicates that the model has a good ability to distinguish between classes. The laundered precision is perfect at 1.0, but the laundered recall is relatively low at 0.32, indicating potential challenges in detecting laundering cases. The true negatives count is 1,897,092, reflecting the number of non-laundering instances correctly identified.
This code creates a bar chart using Plotly to visualize the Gradient-Boosted Trees (GBT) evaluation metrics. It converts the metrics dictionary into a DataFrame and filters out non-numeric values. Each bar represents a metric (such as accuracy, precision, recall) with a corresponding value, and the bars are color-coded. The chart is formatted with the metric values displayed outside the bars, and the y-axis range is set to 0-1.05 for better clarity.
sample_fraud = predictions.filter(col("Is_laundering") == 1).orderBy(rand()).limit(1)
sample_nonfraud = predictions.filter(col("Is_laundering") == 0).orderBy(rand()).limit(1)
sample_combined = sample_fraud.union(sample_nonfraud).orderBy(rand())
extract_prob_udf = udf(lambda v: float(v[1]), DoubleType())
sample_combined = sample_combined.withColumn("fraud_probability", extract_prob_udf(col("probability")))
sample_combined.select(
"Sender_account", "Receiver_account", "Amount", "Payment_type", "Is_Cross_Border",
"Sender_Txn_Count", "Avg_Amount", "Txn_Frequency", "Is_laundering",
"fraud_probability", "prediction"
).show(truncate=False)
Transactions with a higher fraud probability, such as 0.942, are flagged as potential fraud, while lower values (e.g., 0.067) indicate legitimate transactions. The model uses these features to predict fraudulent activity based on transaction patterns. Both predictions were right
This code uses Plotly and Pandas to create a bar chart visualizing the Gradient-Boosted Trees (GBT) evaluation metrics, such as accuracy, precision, recall, and F1 score. First, it converts the metrics_gbt dictionary into a Pandas DataFrame and filters out non-numeric values using .apply(). Then, it creates a bar chart with metrics on the x-axis and corresponding values on the y-axis, applying a random color to each bar. The chart's formatting includes displaying values outside the bars and setting the y-axis range between 0 and 1.05 for better visualization. This is all done within the PySpark and HDFS ecosystem to process large datasets distributed across the cluster, but here, the visualization step is done locally using Pandas for data handling and Plotly for charting.
fig_gbt = px.bar(
pd.DataFrame(list(metrics_gbt.items()), columns=["Metric", "Value"]).loc[
pd.DataFrame(list(metrics_gbt.items()), columns=["Metric", "Value"])["Value"].apply(lambda x: isinstance(x, (int, float)))
],
x="Metric",
y="Value",
text="Value",
color="Metric",
color_discrete_sequence=[f"hsl({random.randint(0, 360)}, 70%, 50%)" for _ in range(len(metrics_gbt))])
fig_gbt.update_traces(texttemplate='%{text:.3f}', textposition='outside')
fig_gbt.update_layout(uniformtext_minsize=8, uniformtext_mode='hide', yaxis_range=[0, 1.05], title="Gradient-Boosted Trees Evaluation Metrics")
fig_gbt.show()
Naive Bayes (NB) is a probabilistic classifier based on Bayes' theorem, which assumes that features are conditionally independent given the class label. The formula for Naive Bayes is:
$$ P(C|X) = \frac{P(C) \prod_{i=1}^{n} P(x_i | C)}{P(X)} $$Where:
The model calculates the posterior probability for each class and assigns the class with the highest probability as the predicted label. The Naive Bayes classifier is effective for problems with many features and is computationally efficient, even with large datasets.
The code assembles the input features (such as 'Hour', 'Day_of_Week', 'Sender_Txn_Count', etc.) into a single vector column called "features" using VectorAssembler. It then scales the features using MinMaxScaler to normalize the data, followed by training a NaiveBayes model on the scaled features, with the target variable being "Is_laundering". After fitting the pipeline on the training data (train_assembled), it makes predictions on the test data (test_assembled).
assembler = VectorAssembler(inputCols=['Hour', 'Day_of_Week', 'Sender_Txn_Count', 'Is_Cross_Border', 'Log_Amount', 'Amount'], outputCol="features")
train_assembled = assembler.transform(train_data)
test_assembled = assembler.transform(test_data)
pipeline = Pipeline(stages=[MinMaxScaler(inputCol="features", outputCol="scaledFeatures"), NaiveBayes(featuresCol="scaledFeatures", labelCol="Is_laundering", smoothing=1.0)])
model = pipeline.fit(train_assembled)
predictions = model.transform(test_assembled)
The evaluate_model_nb function evaluates a Naive Bayes model on metrics such as accuracy, precision, recall, F1 score, and ROC AUC, by calculating True Positives (TP), False Positives (FP), False Negatives (FN), and True Negatives (TN). It uses PySpark's BinaryClassificationEvaluator for ROC AUC and MulticlassClassificationEvaluator for other metrics. The results are printed and displayed in a DataFrame. HDFS is used to store large datasets across distributed nodes for scalability and fault tolerance. PySpark enables distributed data processing and machine learning, leveraging Spark’s distributed computing. These technologies are essential for handling big data and training models efficiently in a scalable manner.
def evaluate_model_nb(predictions):
tp = predictions.filter((col("Is_laundering") == 1) & (col("prediction") == 1)).count()
fp = predictions.filter((col("Is_laundering") == 0) & (col("prediction") == 1)).count()
fn = predictions.filter((col("Is_laundering") == 1) & (col("prediction") == 0)).count()
tn = predictions.filter((col("Is_laundering") == 0) & (col("prediction") == 0)).count()
binary_evaluator = BinaryClassificationEvaluator(labelCol="Is_laundering", rawPredictionCol="rawPrediction", metricName="areaUnderROC")
multi_evaluator = MulticlassClassificationEvaluator(labelCol="Is_laundering", predictionCol="prediction")
return {
"Model": "Naive Bayes", "Accuracy": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "accuracy"}),
"Precision": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "weightedPrecision"}),
"Recall": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "weightedRecall"}),
"F1 Score": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "f1"}),
"ROC AUC": binary_evaluator.evaluate(predictions), "True Negatives": tn
}
metrics_nb = evaluate_model_nb(predictions)
for metric, value in metrics_nb.items():
print("{:<20} {:<10.3f}".format(metric, value) if isinstance(value, (int, float)) else "{:<20} {:<10}".format(metric, value))
nb_results_df = pd.DataFrame(list(metrics_nb.items()), columns=["Metric", "Value"])
display(nb_results_df)
The Naive Bayes model's evaluation metrics show high accuracy (98.9%) and recall (98.9%), indicating the model's effectiveness in identifying fraudulent transactions. Precision is slightly lower at 97.7%, meaning there are some false positives. The F1 Score (98.3%) balances precision and recall, and the ROC AUC of 0.507 indicates moderate model performance in distinguishing between classes. The True Negatives count (1,897,092) indicates how many non-laundering transactions were correctly classified. Overall, the model performs well, but improvements could be made to the ROC AUC and precision.
This code selects a random sample of 3 transactions from the predictions. It then adds a new column "fraud_probability" using a user-defined function (UDF) to extract the fraud probability from the model's raw predictions. The selected columns, including transaction details and the calculated fraud probability, are displayed. Finally, the results are shown with no truncation, displaying full details for the sampled transactions.
sample_combined = predictions.orderBy(rand()).limit(3)
sample_combined = sample_combined.withColumn("fraud_probability", extract_prob_udf(col("rawPrediction")))
sample_combined.select(
"Sender_account", "Receiver_account", "Amount", "Payment_type", "Is_Cross_Border",
"Sender_Txn_Count", "Avg_Amount", "Txn_Frequency", "Is_laundering",
"fraud_probability", "prediction"
).show(truncate=False)
The results show that the model predicts low fraud probability values for all three transactions, with negative raw fraud scores, indicating that these transactions are classified as non-fraudulent. The "prediction" column shows 0, confirming the model's decision of no fraud detected and all are right.
fig_nb = px.bar(
pd.DataFrame(list(metrics_nb.items()), columns=["Metric", "Value"]).loc[
pd.DataFrame(list(metrics_nb.items()), columns=["Metric", "Value"])["Value"].apply(lambda x: isinstance(x, (int, float)))
],
x="Metric",
y="Value",
text="Value",
color="Metric",
color_discrete_sequence=[f"hsl({random.randint(0, 360)}, 70%, 50%)" for _ in range(len(metrics_nb))])
fig_nb.update_traces(texttemplate='%{text:.3f}', textposition='outside')
fig_nb.update_layout(uniformtext_minsize=8, uniformtext_mode='hide', yaxis_range=[0, 1.05], title="Naive Bayes Evaluation Metrics")
fig_nb.show()
Support Vector Machine (SVM) is a supervised machine learning algorithm used for classification and regression tasks. In classification, SVM aims to find the hyperplane that best separates different classes in a feature space, maximizing the margin between the classes.
For a binary classification problem, the SVM model uses the following optimization function to find the optimal hyperplane:
$$ \min_{\mathbf{w}, b} \frac{1}{2} ||\mathbf{w}||^2 $$subject to:
$$ y_i (\mathbf{w}^T \mathbf{x}_i + b) \geq 1, \quad \forall i = 1, 2, \dots, N $$Where:
SVM attempts to find the optimal hyperplane that maximizes the margin (distance between the nearest data points of each class).
This code trains a Linear Support Vector Machine (SVM) model on a dataset, using features from train_assembled and predicting the Is_laundering label. After fitting the model, it evaluates the model’s performance on a test dataset (test_assembled) by calculating key metrics such as accuracy, precision, recall, F1 score, and ROC AUC. The evaluation uses a combination of binary and multiclass classification evaluators. The results are printed in a readable format and stored in a DataFrame for further use or display. The metrics provide insights into how well the model performs in detecting money laundering transactions.
model_name = "Linear SVM"
svm_model = LinearSVC(featuresCol="features", labelCol="Is_laundering", maxIter=10)
fitted_svm = svm_model.fit(train_assembled)
predictions = fitted_svm.transform(test_assembled)
def evaluate_model_svm(predictions):
tp = predictions.filter((col("Is_laundering") == 1) & (col("prediction") == 1)).count()
fp = predictions.filter((col("Is_laundering") == 0) & (col("prediction") == 1)).count()
fn = predictions.filter((col("Is_laundering") == 1) & (col("prediction") == 0)).count()
tn = predictions.filter((col("Is_laundering") == 0) & (col("prediction") == 0)).count()
binary_evaluator = BinaryClassificationEvaluator(labelCol="Is_laundering", rawPredictionCol="rawPrediction", metricName="areaUnderROC")
multi_evaluator = MulticlassClassificationEvaluator(labelCol="Is_laundering", predictionCol="prediction")
metrics = {
"Model": "Linear SVM",
"Accuracy": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "accuracy"}),
"Precision": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "weightedPrecision"}),
"Recall": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "weightedRecall"}),
"F1 Score": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "f1"}),
"ROC AUC": binary_evaluator.evaluate(predictions),
"True Negatives": tn
}
return metrics
metrics_svm = evaluate_model_svm(predictions)
print("\nLinear SVM Results:")
print("{:<20} {:<10}".format('Metric', 'Value'))
print("-" * 30)
for metric, value in metrics_svm.items():
if isinstance(value, (int, float)):
print("{:<20} {:<10.3f}".format(metric, value))
else:
print("{:<20} {:<10}".format(metric, value))
svm_results_df = pd.DataFrame(list(metrics_svm.items()), columns=["Metric", "Value"])
print("\nStored Linear SVM Results in DataFrame:")
display(svm_results_df)
The results show the performance of the Linear SVM model for detecting money laundering transactions. The model achieved an accuracy of 98.9%, indicating it correctly predicted almost 99% of the test samples. Precision of 97.8% suggests that when the model predicts a transaction as laundering. The recall of 98.9% and F1 score of 98.3% balances precision and recall. However, the ROC AUC of 0.66 is relatively low, indicating that the model might not be distinguishing well between classes. True negatives (1,897,092) reflect the number of non-laundering transactions correctly identified.
sample_combined = sample_combined.withColumn("fraud_probability", extract_prob_udf(col("rawPrediction")))
sample_combined.select(
"Sender_account", "Receiver_account", "Amount", "Payment_type", "Is_Cross_Border",
"Sender_Txn_Count", "Avg_Amount", "Txn_Frequency", "Is_laundering",
"fraud_probability", "prediction"
).show(truncate=False)
The results display a sample of 3 transactions with their respective fraud probabilities and predictions. To determine how many predictions are correct, we compare the model's prediction against the actual "Is_laundering" label. Here, all the transactions have a prediction of 0 (non-laundering), and the actual values of "Is_laundering" are also 0 for all transactions. Since the model predicted non-laundering (0) for all transactions and the actual values match, all 3 predictions are correct. Therefore, the accuracy of the model for this sample is 100%.
This code generates a bar chart to visualize the numerical evaluation metrics (such as accuracy, precision, recall, etc.) of the Linear SVM model. It first filters the numeric values from the metrics dataframe, then creates a bar chart using Plotly. Each bar represents a metric, and the color of the bars is randomly assigned using HSL values. The chart displays the metric names on the x-axis and the values on the y-axis, with values shown outside the bars.
df_svm = pd.DataFrame(list(metrics_svm.items()), columns=["Metric", "Value"])
df_svm_numeric = df_svm[df_svm["Value"].apply(lambda x: isinstance(x, (int, float)))]
fig_svm = px.bar(
df_svm_numeric,
x="Metric",
y="Value",
text="Value",
color="Metric",
color_discrete_sequence=[f"hsl({random.randint(0, 360)}, 70%, 50%)" for _ in range(len(df_svm_numeric))]
)
fig_svm.update_traces(texttemplate='%{text:.3f}', textposition='outside')
fig_svm.update_layout(uniformtext_minsize=8, uniformtext_mode='hide', yaxis_range=[0, 1.05], title="Linear SVM Evaluation Metrics")
fig_svm.show()
A Multilayer Perceptron (MLP) is a type of artificial neural network consisting of multiple layers: an input layer, one or more hidden layers, and an output layer. Each layer contains neurons that are connected to neurons in the next layer. MLPs are used for supervised learning tasks like classification or regression.
The formula for a single neuron in the network is:
$$ y = f\left(\sum_{i=1}^n w_i x_i + b\right) $$Where:
The code trains a Multilayer Perceptron (MLP) model to predict money laundering using a dataset. It evaluates the model's performance using several metrics: accuracy, precision, recall, F1 score, ROC AUC, and true negatives. These metrics help assess how well the model distinguishes between laundering and non-laundering transactions. The results are displayed in a readable format and stored in a DataFrame for further analysis. The output includes the MLP model's evaluation results, including the accuracy of 98.9% and other performance measures.
model_name = "Multilayer Perceptron"
mlp_model = MultilayerPerceptronClassifier(
featuresCol="features",
labelCol="Is_laundering",
layers=[len(input_cols), 5, 2],
seed=42
)
predictions = mlp_model.fit(train_assembled).transform(test_assembled)
def evaluate_model_mlp(predictions):
tp = predictions.filter((col("Is_laundering") == 1) & (col("prediction") == 1)).count()
fp = predictions.filter((col("Is_laundering") == 0) & (col("prediction") == 1)).count()
fn = predictions.filter((col("Is_laundering") == 1) & (col("prediction") == 0)).count()
tn = predictions.filter((col("Is_laundering") == 0) & (col("prediction") == 0)).count()
binary_evaluator = BinaryClassificationEvaluator(
labelCol="Is_laundering", rawPredictionCol="rawPrediction", metricName="areaUnderROC"
)
multi_evaluator = MulticlassClassificationEvaluator(
labelCol="Is_laundering", predictionCol="prediction"
)
metrics = {
"Model": "Multilayer Perceptron",
"Accuracy": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "accuracy"}),
"Precision": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "weightedPrecision"}),
"Recall": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "weightedRecall"}),
"F1 Score": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "f1"}),
"ROC AUC": binary_evaluator.evaluate(predictions),
"True Negatives": tn
}
return metrics
metrics_mlp = evaluate_model_mlp(predictions)
print("\nMultilayer Perceptron Results:")
print("{:<20} {:<10}".format('Metric', 'Value'))
print("-" * 30)
for metric, value in metrics_mlp.items():
if isinstance(value, (int, float)):
print("{:<20} {:<10.3f}".format(metric, value))
else:
print("{:<20} {:<10}".format(metric, value))
mlp_results_df = pd.DataFrame(list(metrics_mlp.items()), columns=["Metric", "Value"])
print("\nStored MLP Results in DataFrame:")
display(mlp_results_df)
The Multilayer Perceptron (MLP) model achieved strong performance with an accuracy of 98.9%, indicating a high proportion of correct predictions. Precision and recall are also high, at 97.8% and 98.9%, respectively, showing the model is good at both identifying fraudulent transactions and minimizing false positives. The F1 score of 98.3% suggests a balanced performance between precision and recall. The ROC AUC of 0.58 indicates moderate model performance in distinguishing between classes. The model correctly identified 1,897,092 true negatives, implying it accurately classified non-laundering transactions.
The code randomly selects 3 samples from the predictions dataset and adds a new column called "fraud_score" that contains the raw prediction values. It then selects specific columns from the dataset, including sender and receiver information, transaction details, and predicted values, and displays the results in a readable format. The show(truncate=False) ensures that the full content of the selected columns is displayed without truncation.
sample_combined = predictions.orderBy(rand()).limit(3)
sample_combined = sample_combined.withColumn("fraud_score", col("rawPrediction"))
sample_combined.select(
"Sender_account", "Receiver_account", "Amount", "Payment_type", "Is_Cross_Border",
"Sender_Txn_Count", "Avg_Amount", "Txn_Frequency", "Is_laundering",
"fraud_score", "prediction"
).show(truncate=False)
The "fraud_score" is an array with two values, likely representing the model's raw prediction outputs for each class. The "prediction" column indicates whether the transaction is flagged as fraudulent (1) or not (0). Based on the displayed results, all three predictions are 0.0, meaning the model has classified these transactions as non-fraudulent, and none of the predictions are flagged as correct for fraud.
metrics_to_plot = mlp_results_df[mlp_results_df['Metric'].isin(['Accuracy', 'Precision', 'Recall', 'F1 Score', 'ROC AUC'])]
fig = px.bar(metrics_to_plot, x="Metric", y="Value", title="Multilayer Perceptron Model Metrics",
labels={"Metric": "Evaluation Metric", "Value": "Metric Value"}, color="Metric",
color_discrete_map={'Accuracy': 'blue', 'Precision': 'green', 'Recall': 'red', 'F1 Score': 'orange', 'ROC AUC': 'purple'})
fig.show()
Here, we are comparing the performance of multiple machine learning models(Reference -4). First, we create a list that holds the name of each model and its evaluation results (stored earlier in different DataFrames). Then, we define a list of key metrics like Accuracy, Precision, Recall, F1 Score, and ROC AUC that we want to compare across all models. We create an empty table with these metrics as columns. For each model, we fetch the corresponding metric values and add them into the table. If a metric is missing for a model, we leave it blank (None). After that, we format all numeric values to show only three decimal places for neatness. Finally, we display the full comparison table to easily see which model performed better on different metrics.
Accuracy
Meaning: How often the model predicted correctly overall.
Formula:
$$
\text{Accuracy} = \frac{\text{TP} + \text{TN}}{\text{TP} + \text{TN} + \text{FP} + \text{FN}}
$$
Precision
Meaning: When the model predicted laundering (positive class), how often was it correct?
Formula:
$$
\text{Precision} = \frac{\text{TP}}{\text{TP} + \text{FP}}
$$
Recall (Sensitivity)
Meaning: Of all actual laundering cases, how many did the model correctly identify?
Formula:
$$
\text{Recall} = \frac{\text{TP}}{\text{TP} + \text{FN}}
$$
F1 Score
Meaning: Balance between Precision and Recall, especially important for imbalanced datasets.
Formula:
$$
\text{F1 Score} = 2 \times \frac{\text{Precision} \times \text{Recall}}{\text{Precision} + \text{Recall}}
$$
ROC AUC (Area Under the ROC Curve)
Meaning: Measures the model's ability to distinguish between classes.
Laundered Precision
Meaning: Precision specifically for laundering transactions (positive class).
Formula:
$$
\text{Laundered Precision} = \frac{\text{TP}}{\text{TP} + \text{FP}}
$$
Laundered Recall
Meaning: Recall specifically for laundering transactions (positive class).
Formula:
$$
\text{Laundered Recall} = \frac{\text{TP}}{\text{TP} + \text{FN}}
$$
True Negatives
Meaning: The number of times the model correctly predicted a non-laundering (normal) transaction.
(No formula needed — it is a direct count of TN.)
Important Terms:
model_dfs = [
("Logistic Regression", lr_results_df),
("Decision Tree", dt_results_df),
("Random Forest", rf_results_df),
("Gradient-Boosted Trees", gbt_results_df),
("Naive Bayes", nb_results_df),
("Linear SVM", svm_results_df),
("Multilayer Perceptron", mlp_results_df)
]
metrics_list = [
"Accuracy", "Precision", "Recall", "F1 Score", "ROC AUC",
"Laundered Precision", "Laundered Recall", "True Negatives"
]
comparison_table = pd.DataFrame(columns=["Model"] + metrics_list)
for model_name, df in model_dfs:
model_metrics = {"Model": model_name}
for metric in metrics_list:
if metric in df["Metric"].values:
value = df[df["Metric"] == metric]["Value"].iloc[0]
model_metrics[metric] = value
else:
model_metrics[metric] = None
comparison_table = comparison_table.append(model_metrics, ignore_index=True)
numeric_columns = ["Accuracy", "Precision", "Recall", "F1 Score", "ROC AUC", "Laundered Precision", "Laundered Recall"]
for col in numeric_columns:
comparison_table[col] = comparison_table[col].apply(lambda x: f"{x:.3f}" if pd.notnull(x) else "N/A")
print("\nModel Comparison Table:")
display(comparison_table)
The table compares seven different machine learning models based on how well they identify laundering transactions (like catching illegal money transfers) versus normal ones. Each model is evaluated using metrics like Accuracy, Precision, Recall, F1 Score, ROC AUC, Laundered Precision, Laundered Recall, and True Negatives. These metrics tell us how often the model is correct overall, how good it is at spotting laundering cases without mistakes, and how well it avoids missing actual laundering cases. For example, Accuracy shows the percentage of all predictions (laundering or not) that are correct, while Laundered Precision and Recall focus specifically on how well the model handles laundering cases. True Negatives count how many normal transactions the model correctly identifies as non-laundering.
From the table, the Decision Tree model stands out with the highest scores across most metrics (e.g., 0.999 for Accuracy, Precision, Recall, and F1 Score), meaning it’s very reliable at both catching laundering and avoiding errors. Logistic Regression also performs well, especially with a perfect ROC AUC of 1.000, indicating it’s excellent at separating laundering from non-laundering transactions. However, models like Naive Bayes, Linear SVM, and Multilayer Perceptron have lower scores, especially in ROC AUC (around 0.5–0.67), suggesting they’re not much better than random guessing at distinguishing between classes.
The code first reshapes the comparison table into a long format using the melt function, where each row represents a model-metric pair. It then ensures that all metric values are numeric and filters out the 'True Negatives' metric since it does not represent a rate. After that, it creates a grouped bar chart using Plotly Express to visualize how each model performs across different metrics. Finally, it displays the interactive chart, making it easy to compare the strengths and weaknesses of each model side-by-side.
comparison_melted = comparison_table.melt(id_vars=["Model"], value_vars=metrics_list, var_name="Metric", value_name="Value")
comparison_melted['Value'] = pd.to_numeric(comparison_melted['Value'], errors='coerce')
comparison_melted = comparison_melted[comparison_melted['Metric'] != 'True Negatives']
fig = px.bar(
comparison_melted,
x="Model",
y="Value",
color="Metric",
barmode="group",
title="Model Comparison by Metrics",
labels={"Value": "Metric Value", "Model": "Models", "Metric": "Metrics"},
)
fig.show()
We trained the first model to detect money laundering, but even though the results were good, there was still some room for improvement. To make the model smarter and more accurate, we retrained it by changing a few settings to help it learn better. By doing this, the model could catch more laundering cases without making too many mistakes. Retraining helps the model understand the data more deeply and make better decisions. The goal was to build the best possible model for detecting laundering, not just a good one. After retraining, we checked the new model’s results and found it was even stronger and more reliable. That’s why retraining is important — it helps us get the best performance for real-world use. So, I decided to retrain the Decision Tree by adjusting its settings, like making it deeper and letting it split more smartly (Reference -5). Retraining helps the tree learn more patterns in the data and make better, more accurate predictions. In short, we retrain the Decision Tree to make it even smarter at catching money laundering activities.
The PySpark code defines a function evaluate_model that calculates performance metrics for a binary classification model, likely used to detect laundering transactions, with data stored in HDFS. It takes a DataFrame predictions with columns Is_laundering (true labels) and prediction (model predictions). The function computes True Positives (TP), False Positives (FP), False Negatives (FN), and True Negatives (TN) by filtering the DataFrame using PySpark’s filter and count methods, leveraging HDFS for distributed storage and processing. It uses BinaryClassificationEvaluator to calculate the ROC AUC and MulticlassClassificationEvaluator for Accuracy, Precision, Recall, and F1 Score, accessing data from HDFS via PySpark’s distributed computing. Additionally, it manually computes Laundered Precision (TP/(TP+FP)) and Laundered Recall (TP/(TP+FN)) for the positive class (laundering), handling edge cases where denominators are zero. The metrics are returned as a dictionary. Since the data is processed in a distributed manner on HDFS, PySpark ensures scalability for large datasets, common in financial transaction analysis.
def evaluate_model(predictions):
tp = predictions.filter((F.col("Is_laundering") == 1) & (F.col("prediction") == 1)).count()
fp = predictions.filter((F.col("Is_laundering") == 0) & (F.col("prediction") == 1)).count()
fn = predictions.filter((F.col("Is_laundering") == 1) & (F.col("prediction") == 0)).count()
tn = predictions.filter((F.col("Is_laundering") == 0) & (F.col("prediction") == 0)).count()
binary_evaluator = BinaryClassificationEvaluator(
labelCol="Is_laundering", rawPredictionCol="rawPrediction", metricName="areaUnderROC"
)
multi_evaluator = MulticlassClassificationEvaluator(
labelCol="Is_laundering", predictionCol="prediction"
)
metrics = {
"Accuracy": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "accuracy"}),
"Precision": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "weightedPrecision"}),
"Recall": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "weightedRecall"}),
"F1 Score": multi_evaluator.evaluate(predictions, {multi_evaluator.metricName: "f1"}),
"ROC AUC": binary_evaluator.evaluate(predictions),
"Laundered Precision": tp / (tp + fp) if (tp + fp) > 0 else 0,
"Laundered Recall": tp / (tp + fn) if (tp + fn) > 0 else 0,
"True Negatives": tn
}
return metrics
The confusion_matrix_spark function calculates the number of true positives, true negatives, false positives, and false negatives from the model’s predictions. It then organizes these results into a table to clearly show how well the model classified the data. The plot_confusion_matrix function takes this table and creates a heatmap using Plotly to visually display the confusion matrix. The x-axis of the heatmap represents predicted labels while the y-axis shows the actual labels. This visual representation helps quickly understand the model’s performance at a glance.
def confusion_matrix_spark(predictions):
tp = predictions.filter((F.col("Is_laundering") == 1) & (F.col("prediction") == 1)).count()
fp = predictions.filter((F.col("Is_laundering") == 0) & (F.col("prediction") == 1)).count()
fn = predictions.filter((F.col("Is_laundering") == 1) & (F.col("prediction") == 0)).count()
tn = predictions.filter((F.col("Is_laundering") == 0) & (F.col("prediction") == 0)).count()
return pd.DataFrame(
[[tn, fp], [fn, tp]],
columns=["Predicted No", "Predicted Yes"],
index=["Actual No", "Actual Yes"]
)
def plot_confusion_matrix(conf_matrix_df, model_name):
fig = go.Figure(data=go.Heatmap(
z=conf_matrix_df.values,
x=conf_matrix_df.columns,
y=conf_matrix_df.index,
colorscale='Blues',
colorbar=dict(title="Count"),
text=conf_matrix_df.values.astype(str), # Adding text annotations
hoverinfo='text'
))
fig.update_layout(
title=f"Confusion Matrix: {model_name}",
xaxis=dict(title="Predicted", side="top"),
yaxis=dict(title="Actual")
)
fig.show()
This code prepares both numeric and categorical data by filling missing values and indexing categories. It builds a machine learning pipeline that includes feature assembling, scaling, and training a Decision Tree classifier. First, the model is trained with initial hyperparameters (like max depth 10 and impurity "gini") and evaluated. Then, the Decision Tree is retrained with new hyperparameters (like deeper trees, more nodes, and impurity "entropy") to try improving performance. The model is re-evaluated, and new results are stored in a DataFrame. Finally, a confusion matrix is created and plotted to visually check how well the retrained model predicts. This helps compare the initial and retrained models side by side.
categorical_columns = ["Payment_currency", "Received_currency", "Sender_bank_location", "Receiver_bank_location", "Payment_type"]
numeric_features = ["Sender_Txn_Count", "Is_Cross_Border", "Log_Amount", "Amount", "Hour", "Day_of_Week", "Avg_Amount", "Txn_Frequency"]
assembler_inputs = numeric_features + [f"{col_name}_indexed" for col_name in categorical_columns]
for col_name in numeric_features:
train_data = train_data.fillna({col_name: 0})
test_data = test_data.fillna({col_name: 0})
for col_name in categorical_columns:
train_data = train_data.fillna({col_name: "Unknown"})
test_data = test_data.fillna({col_name: "Unknown"})
indexers = [StringIndexer(inputCol=col_name, outputCol=f"{col_name}_indexed", handleInvalid="keep") for col_name in categorical_columns]
assembler = VectorAssembler(inputCols=assembler_inputs, outputCol="features", handleInvalid="keep")
scaler = StandardScaler(inputCol="features", outputCol="scaled_features", withStd=True, withMean=True)
dt_model = DecisionTreeClassifier(
featuresCol="scaled_features",
labelCol="Is_laundering",
maxDepth=10,
minInstancesPerNode=50,
impurity="gini",
seed=42
)
pipeline = Pipeline(stages=indexers + [assembler, scaler, dt_model])
model = pipeline.fit(train_data)
predictions = model.transform(test_data)
metrics = evaluate_model(predictions)
print("\nInitial Decision Tree Results:")
print("{:<20} {:<10}".format('Metric', 'Value'))
print("-" * 30)
for metric, value in metrics.items():
if isinstance(value, (int, float)):
print("{:<20} {:<10.3f}".format(metric, value))
else:
print("{:<20} {:<10}".format(metric, value))
dt_results_df = pd.DataFrame(list(metrics.items()), columns=["Metric", "Value"])
dt_model_retrained = DecisionTreeClassifier(
featuresCol="scaled_features",
labelCol="Is_laundering",
maxDepth=15,
minInstancesPerNode=100,
maxBins=32,
impurity="entropy",
minInfoGain=0.01,
seed=42
)
pipeline_retrained = Pipeline(stages=indexers + [assembler, scaler, dt_model_retrained])
model_retrained = pipeline_retrained.fit(train_data)
predictions_retrained = model_retrained.transform(test_data)
metrics_retrained = evaluate_model(predictions_retrained)
print("\nRetrained Decision Tree Results with New Hyperparameters:")
print("{:<20} {:<10}".format('Metric', 'Value'))
print("-" * 30)
for metric, value in metrics_retrained.items():
if isinstance(value, (int, float)):
print("{:<20} {:<10.3f}".format(metric, value))
else:
print("{:<20} {:<10}".format(metric, value))
dt_results_retrained_df = pd.DataFrame(list(metrics_retrained.items()), columns=["Metric", "Value"])
print("\nStored Retrained Decision Tree Results in DataFrame:")
display(dt_results_retrained_df)
conf_matrix_retrained = confusion_matrix_spark(predictions_retrained)
plot_confusion_matrix(conf_matrix_retrained, "Retrained Decision Tree")
The initial decision tree performed exceptionally well, showing very high accuracy, precision, recall, and F1 score, all close to 0.999. It also had a strong ROC AUC of 0.964, indicating good ability to distinguish between fraud and non-fraud cases. Laundered precision was nearly perfect at 0.999, and laundered recall was strong at 0.919, meaning the model caught most laundering cases while keeping false positives low. It correctly identified 1,897,070 non-fraud cases.
After tuning the model with new hyperparameters, the overall accuracy and precision remained high. The laundered precision even improved to a perfect 1.000, meaning it avoided false positives entirely for laundering. However, the laundered recall slightly dropped to 0.906, indicating it missed more laundering cases than before. The ROC AUC also slightly decreased to 0.953. Although the number of true negatives slightly increased to 1,897,092, the trade-off between better precision and lower recall may not be ideal depending on the use case.
The development of a scalable anti-money laundering (AML) detection system using Apache Spark and HDFS faced several technical challenges due to the dataset's size, imbalance, and computational demands. To address these, strategic optimizations, such as implementing SMOTE for class balancing, fine-tuning Spark configurations for scalability, and applying efficient data processing techniques, were employed. These efforts ensured the system's ability to process 9.5 million records effectively and deliver accurate fraud detection.
The development of the scalable anti-money laundering (AML) detection system using Apache Spark and HDFS encountered several technical challenges, each requiring tailored solutions to ensure efficiency and accuracy. These challenges, rooted in the dataset’s size, imbalance, and computational demands, were addressed through strategic optimizations and innovative techniques, enabling the system to process 9.5 million records effectively.
Class Imbalance: The dataset’s severe class imbalance, with only 0.1% (9,873) fraud cases against 9,494,979 non-fraud records, posed a risk of models overfitting to the majority class, leading to poor fraud detection. To address this, a custom Synthetic Minority Oversampling Technique (SMOTE) was implemented in PySpark, generating synthetic fraud samples to increase the fraud class to 108,603 records. This balanced the dataset, enhancing model sensitivity to fraud. Careful tuning of SMOTE’s parameters, such as the number of nearest neighbors, was critical to minimize noise and prevent artificial patterns from skewing results, ensuring robust model performance.
Scalability: Processing a 996 MB dataset with 9.5 million records demanded efficient distributed computing. HDFS provided fault-tolerant, distributed storage, splitting data across nodes for parallel access and reliability. Spark’s DataFrame API, powered by the Catalyst optimizer, enabled fast, parallel processing of large-scale operations. However, initial runs on local clusters faced memory bottlenecks, causing executor failures. This was resolved by optimizing Spark configurations, including increasing executor memory (e.g., spark.executor.memory to 4GB) and adjusting partition sizes to balance workloads, ensuring seamless scalability for production-level data volumes.
Feature Engineering: Deriving risk indicators like Sender_Txn_Count and Is_Cross_Border required complex PySpark Window functions to aggregate transaction counts and identify cross-border patterns. These operations were computationally intensive, slowing down the pipeline. To mitigate this, intermediate DataFrames were cached using df.cache() to reduce redundant computations, and join operations were optimized by broadcasting smaller tables, minimizing shuffle costs. These strategies significantly improved processing speed while maintaining feature quality.
Model Selection and Tuning: Evaluating seven models—Logistic Regression, Decision Tree, Random Forest, Gradient-Boosted Trees, Naive Bayes, Linear SVM, and Multilayer Perceptron—required extensive testing to balance accuracy, precision, and recall. The Decision Tree initially achieved high accuracy (0.999). Retraining with tuned hyperparameters (maxDepth=15, impurity=entropy) improved laundered precision to 1.0, eliminating false positives with a slightly lower ROC-AUC.
Visualization: Converting large Spark DataFrames to Pandas for Plotly visualizations risked memory overload on local machines. To address this, data was sampled or aggregated (e.g., grouping by payment type) before conversion, reducing memory usage while preserving insights. This ensured scalable, effective visualizations like confusion matrices and performance bar charts. These solutions collectively enabled a robust, scalable AML detection system, overcoming challenges through optimized configurations, advanced techniques like SMOTE, and efficient data handling, paving the way for reliable fraud detection in large-scale financial datasets.
The project, "Scalable Anti-Money Laundering Detection Using Apache Spark and HDFS," successfully built a strong and scalable system to spot money laundering in a 996 MB dataset with 9.5 million transaction records. By using big data tools—Apache Spark for fast processing and HDFS for safe storage—the project met its goal of creating an accurate and compliant anti-money laundering (AML) system. Since the code is executed on a remote cluster node (e.g., dsm10 at Goldsmiths College), the Spark driver and executors are running on the university’s distributed Hadoop cluster. This ensures that computation is both powerful and scalable, capable of processing large volumes of transactional data in parallel.
The retrained Decision Tree model became the best performer, showing almost perfect results with 99.9% accuracy, 1.0 laundered precision, and 0.90 laundered recall. The high precision means very few false alarms, which is important for following Financial Action Task Force (FATF) rules (Reference -6), and the recall shows good fraud catching, though there’s still room to improve.
Handling technical problems was key to the project’s success. A big class imbalance (only 0.1% fraud) was fixed using a custom SMOTE method in PySpark, which created fake fraud examples to balance the data. Scalability was achieved through HDFS’s shared storage and Spark’s fast DataFrame tools, with tweaks like more executor memory solving memory problems. Feature engineering, like adding Sender_Txn_Count and Is_Cross_Border, made the model stronger, while tuning the model settings (like maxDepth=15 and using entropy) helped balance precision and recall.Even with the great results, there is still room to improve. Testing the model on different real-world datasets would also check if the model works well outside the current data, because using synthetic SMOTE data can sometimes create patterns that don’t happen in real life.
This project moves forward big data research in financial crime detection, giving banks and financial companies a strong, scalable plan for AML systems. Built on fast distributed computing and smart machine learning, this framework is ready for the growing number of digital transactions and gives a strong starting point for fighting money laundering.
The "Scalable Anti-Money Laundering Detection Using Apache Spark and HDFS" project demonstrated how big data technologies can be leveraged to combat money laundering. By processing a 9.5 million-record dataset with Apache Spark and HDFS, the system became scalable and reliable, making it suitable for real-time financial transactions. The retrained Decision Tree model, with 99.9% accuracy and 1.0 laundered precision, effectively balanced precision and recall, helping meet Financial Action Task Force (FATF) standards by reducing false positives. Features like Sender_Txn_Count and Is_Cross_Border played a key role in detecting fraud, confirming their importance as risk indicators. Using SMOTE to address the 0.1% fraud class imbalance improved the model’s ability to detect fraud, but the use of synthetic data could lead to overfitting risks. The project’s solutions for visualization and scalability, like sampling for Plotly and optimizing Spark configurations, made it easier to deploy but showed the need for strong infrastructure in real-world settings. The YARN cluster manager was utilized with HDFS, ensuring scalability and a reliable system architecture for handling large datasets in real-time.
Future work should focus on improving recall by using ensemble methods that combine Decision Tree and Logistic Regression to catch more fraud while keeping precision high. Testing the model on different real-world datasets will help address the limits of SMOTE and make the system more reliable. In addition, integrating real-time streaming with Spark Streaming could allow for dynamic fraud detection. A feature store, like Hopsworks, provides a centralized repository for storing and managing precomputed features for machine learning models. It integrates with HDFS and PySpark to streamline feature engineering and model training. Exploring advanced techniques like deep learning or graph-based anomaly detection might improve accuracy, positioning the system as an advanced solution for AML in today’s digital world.